From c1ce5f34ad3fbedbd4e50fb5d85c2783fcbcacd6 Mon Sep 17 00:00:00 2001 From: andrewn6 <59238070+andrewn6@users.noreply.github.com> Date: Sat, 14 Nov 2026 17:45:38 -0500 Subject: [PATCH 1/6] feat: plan artifact moves as journal-safe jobs with retry-safe steps and a durable mover hook --- internal/orchestrator/artifacts.go | 122 ++--- internal/orchestrator/artifacts_durable.go | 418 ++++++++++++++++++ .../orchestrator/artifacts_durable_test.go | 303 +++++++++++++ internal/orchestrator/delta.go | 76 +--- internal/orchestrator/fleet.go | 4 + internal/orchestrator/migrate.go | 17 +- 6 files changed, 776 insertions(+), 164 deletions(-) create mode 100644 internal/orchestrator/artifacts_durable.go create mode 100644 internal/orchestrator/artifacts_durable_test.go diff --git a/internal/orchestrator/artifacts.go b/internal/orchestrator/artifacts.go index d7f49cd5..9e3819d5 100644 --- a/internal/orchestrator/artifacts.go +++ b/internal/orchestrator/artifacts.go @@ -535,7 +535,7 @@ func (fm *FleetManager) ensureArtifactOnHost(ctx context.Context, record Snapsho return holder.SnapshotID, nil } - source, target, err := fm.artifactEndpoints(record, holders, hostID) + source, _, err := fm.artifactEndpoints(record, holders, hostID) if err != nil { return "", err } @@ -549,113 +549,37 @@ func (fm *FleetManager) ensureArtifactOnHost(ctx context.Context, record Snapsho localID = liveReplicaSnapshotID(record.SnapshotID, hostID) } - // Both ends are held for the duration: the source is being read and the - // destination is being written, and collecting either mid-flight leaves a - // torn rootfs that a later create would happily boot. - fm.beginArtifactPull(record.SnapshotID, source.snapshotID, localID) - defer fm.endArtifactPull(record.SnapshotID, source.snapshotID, localID) - - pullCtx := ctx - if fm.artifactPullTimeout > 0 { - var cancel context.CancelFunc - pullCtx, cancel = context.WithTimeout(ctx, fm.artifactPullTimeout) - defer cancel() - } - fm.logger.Info("copying artifact between hosts", "digest", record.Digest, "from", source.endpoint.HostID, "to", hostID, "snapshot", localID) - moved, err := fm.artifactMover.MoveArtifact(pullCtx, ArtifactMove{ - Digest: record.Digest, - SnapshotID: localID, - From: source.endpoint, - To: target, - Files: files, - Lazy: lazy, - + // Both ends are held for the duration: the source is being read and the + // destination is being written, and collecting either mid-flight leaves a + // torn rootfs that a later create would happily boot. + job := ArtifactMoveJob{ + SnapshotID: record.SnapshotID, + Digest: record.Digest, + SourceHostID: source.endpoint.HostID, SourceSnapshotID: source.snapshotID, - }) - if err == nil && record.Kind == SnapshotKindLive && moved.Kind != SnapshotKindLive { - // the target took the rootfs and dropped the rest. nothing is recorded, - // so nothing can seed from the half it committed. - err = fmt.Errorf("host %s committed only the rootfs of a live snapshot; its host agent cannot receive memory images", hostID) - } - if err != nil { + HostID: hostID, + LocalID: localID, + Files: files, + Lazy: lazy, + TenantID: record.TenantID, + LayerKey: record.LayerKey, + Arch: record.Arch, + Kind: record.Kind, + SizeBytes: record.SizeBytes, + } + return fm.runArtifactMove(ctx, job, func(err error) error { // Cancellation is reported as itself rather than as a transfer failure: // the caller gave up, the peer did nothing wrong, and a dead-lettered // "transfer failed" would send an operator looking at the network. if ctxErr := ctx.Err(); ctxErr != nil { - return "", fmt.Errorf("copy artifact %s to host %s: %w", record.Digest, hostID, ctxErr) + return fmt.Errorf("copy artifact %s to host %s: %w", record.Digest, hostID, ctxErr) } - return "", fmt.Errorf("%w: copy artifact %s from host %s to host %s: %w", + return fmt.Errorf("%w: copy artifact %s from host %s to host %s: %w", ErrArtifactTransferFailed, record.Digest, source.endpoint.HostID, hostID, err) - } - - committedID := moved.SnapshotID - if committedID == "" { - committedID = localID - } - if lazy { - // the target holds no memory image, so it can never be the source of - // the next move or a seed for anything else, and its agent deletes the - // seed once the vm has resumed. recording it would say otherwise. - return committedID, nil - } - size := moved.SizeBytes - if size <= 0 { - size = record.SizeBytes - } - - now := time.Now() - replicaMetadata := map[string]string{ - "source_snapshot_id": record.SnapshotID, - "source_host_id": source.endpoint.HostID, - } - if len(files) > 0 { - // the copy can be a source for the next move, so it carries the digests. - replicaMetadata = withLiveFiles(replicaMetadata, files) - } - metadata, _ := marshalSnapshotMetadata("", replicaMetadata) - replica := SnapshotRecord{ - SnapshotID: committedID, - // No VM: this copy was not checkpointed from anything running here, it - // arrived over the wire. Leaving VMID empty is what marks it as a - // free-standing artifact rather than a vm's checkpoint. - VMID: "", - HostID: hostID, - TenantID: record.TenantID, - // ParentSnapshotID is deliberately NOT the source. Lineage means "this - // snapshot descends from that one", and deleting a snapshot with - // children is refused; pointing at the source would make every origin - // artifact undeletable the moment anything pulled a copy of it. - Mode: SnapshotModeBuild, - LayerKey: record.LayerKey, - Arch: record.Arch, - Digest: record.Digest, - Kind: record.Kind, - State: SnapshotStateReady, - SizeBytes: size, - Metadata: metadata, - CreatedAt: now, - UpdatedAt: now, - } - if err := fm.upsertSnapshotRecord(ctx, replica); err != nil { - // The bytes are on the host but nothing records them. Failing is the - // consistent choice: the create that triggered this would fail at its - // own next persist anyway, and the id is derived from (digest, host) so - // the next attempt re-commits over the same directory rather than - // leaving one orphan per try. - return "", fmt.Errorf("persist pulled artifact %s on host %s: %w", committedID, hostID, err) - } - - fm.appendEventBackground("host", hostID, "artifact.pulled", map[string]any{ - "digest": record.Digest, - "snapshot_id": committedID, - "source_host_id": source.endpoint.HostID, - "source_snapshot_id": record.SnapshotID, - "size_bytes": size, }) - return committedID, nil } // artifactSource is a chosen peer to pull from, plus the id it holds the @@ -837,6 +761,10 @@ func (fm *FleetManager) reconcileArtifacts(ctx context.Context) { collect = append(collect, overflowArtifacts(candidates, protected, fm.artifactMaxPerTenant)...) } + // a move running on restate holds its pins there, where they outlive a + // restart of this process. + collect = fm.dropDurablyPinned(ctx, collect) + for _, record := range collect { fm.evictArtifact(ctx, record) } diff --git a/internal/orchestrator/artifacts_durable.go b/internal/orchestrator/artifacts_durable.go new file mode 100644 index 00000000..24e17770 --- /dev/null +++ b/internal/orchestrator/artifacts_durable.go @@ -0,0 +1,418 @@ +package orchestrator + +import ( + "context" + "errors" + "fmt" + "net/http" + "time" +) + +// ArtifactMoveJob is one planned host-to-host artifact move, and the input of +// the durable Artifacts/{local_id}.Move handler. it names hosts by id only: +// their agent tokens are looked up inside the pull step, so no secret reaches +// the restate journal. +type ArtifactMoveJob struct { + // SnapshotID is the record being moved, Digest its rootfs digest. + SnapshotID string `json:"snapshot_id"` + Digest string `json:"digest"` + // SourceHostID serves the bytes, stored there as SourceSnapshotID. + SourceHostID string `json:"source_host_id"` + SourceSnapshotID string `json:"source_snapshot_id"` + // HostID receives them and commits them as LocalID. + HostID string `json:"host_id"` + LocalID string `json:"local_id"` + + Files map[string]string `json:"files,omitempty"` + Lazy bool `json:"lazy,omitempty"` + // Base makes this a delta move merged onto Base on HostID, see + // moveDeltaToHost. + Base string `json:"base,omitempty"` + DropBase bool `json:"drop_base,omitempty"` + + // what the copy's record takes from the moved record. + TenantID string `json:"tenant_id"` + LayerKey string `json:"layer_key,omitempty"` + Arch string `json:"arch,omitempty"` + Kind SnapshotKind `json:"kind,omitempty"` + SizeBytes int64 `json:"size_bytes,omitempty"` +} + +// PinIDs is every snapshot id the move reads or writes, which gc must leave +// alone while it runs. +func (j ArtifactMoveJob) PinIDs() []string { + ids := []string{j.SnapshotID} + for _, id := range []string{j.SourceSnapshotID, j.LocalID, j.Base} { + if id != "" && !containsString(ids, id) { + ids = append(ids, id) + } + } + return ids +} + +func containsString(list []string, s string) bool { + for _, v := range list { + if v == s { + return true + } + } + return false +} + +// DurableArtifacts runs artifact moves as the restate Artifacts/{local_id}.Move +// handler and keeps pull pins in restate state, so neither is lost when the +// orchestrator restarts. *durable.Client satisfies it. +type DurableArtifacts interface { + // MoveArtifact runs the move and waits for it, returning the snapshot id + // committed on the destination. the move keeps running if ctx ends first. + MoveArtifact(ctx context.Context, job ArtifactMoveJob) (string, error) + // PinArtifacts holds ids against gc for holder until ReleaseArtifacts, or + // until ttl passes. + PinArtifacts(ctx context.Context, holder string, ttl time.Duration, ids ...string) error + ReleaseArtifacts(ctx context.Context, holder string, ids ...string) error + // ArtifactPinned reports whether anything holds a pin on id. + ArtifactPinned(ctx context.Context, id string) (bool, error) +} + +// ArtifactMoveMetrics is an optional extension of ReconcileMetrics, recording +// each attempt at a host-to-host artifact copy. +type ArtifactMoveMetrics interface { + ArtifactMoveCompleted(sourceHostID, targetHostID string, bytes int64, d time.Duration, err error) +} + +// SetDurableArtifacts routes artifact moves and pull pins through restate. nil +// (mode off, the default) keeps them in process. set it before Start. +func (fm *FleetManager) SetDurableArtifacts(d DurableArtifacts) { + fm.mu.Lock() + fm.durableArtifacts = d + fm.mu.Unlock() +} + +func (fm *FleetManager) durableArtifactsClient() DurableArtifacts { + fm.mu.RLock() + defer fm.mu.RUnlock() + return fm.durableArtifacts +} + +// durableMoveWait bounds how long a caller with no deadline of its own waits +// on a durable move. +const durableMoveWait = time.Hour + +// runArtifactMove carries out a planned move: on restate when it is +// configured, otherwise in process holding in-memory pins. wrap turns a failed +// copy into the caller's error; a failed record write in process is returned +// as itself, as before. +func (fm *FleetManager) runArtifactMove(ctx context.Context, job ArtifactMoveJob, wrap func(error) error) (string, error) { + if d := fm.durableArtifactsClient(); d != nil { + waitCtx := ctx + if _, ok := ctx.Deadline(); !ok { + // a background caller (the checkpoint loop) would otherwise wait + // forever on a restate that accepts and never answers. giving up + // leaves the move running; its copy is found on the next try. + var cancel context.CancelFunc + waitCtx, cancel = context.WithTimeout(ctx, durableMoveWait) + defer cancel() + } + id, err := d.MoveArtifact(waitCtx, job) + if err != nil { + return "", wrap(err) + } + // a fresh use keeps the copy out of the next sweep, as the in-process + // release does. + fm.touchArtifact(job.PinIDs()...) + return id, nil + } + ids := job.PinIDs() + fm.beginArtifactPull(ids...) + defer fm.endArtifactPull(ids...) + moved, err := fm.ArtifactPull(ctx, job) + if err != nil { + return "", wrap(err) + } + return fm.ArtifactRecord(ctx, job, moved) +} + +// holdArtifact pins id against gc for holder and returns its release. on +// restate the pin outlives a restart and lapses after ttl if nothing releases +// it; a pin restate refuses falls back to an in-memory one. +func (fm *FleetManager) holdArtifact(ctx context.Context, holder string, ttl time.Duration, id string) (release func()) { + if d := fm.durableArtifactsClient(); d != nil { + if err := d.PinArtifacts(ctx, holder, ttl, id); err == nil { + return func() { + rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := d.ReleaseArtifacts(rctx, holder, id); err != nil { + fm.logger.Warn("release durable artifact pin failed; it lapses on its timer", "holder", holder, "snapshot", id, "err", err) + } + } + } else { + fm.logger.Warn("durable artifact pin failed; holding it in memory", "holder", holder, "snapshot", id, "err", err) + } + } + fm.beginArtifactPull(id) + return func() { fm.endArtifactPull(id) } +} + +// dropDurablyPinned removes from collect every artifact a durable pin holds. +// a pin that cannot be read counts as held: gc waits for the next sweep +// rather than deleting something a move may be writing. +func (fm *FleetManager) dropDurablyPinned(ctx context.Context, collect []SnapshotRecord) []SnapshotRecord { + d := fm.durableArtifactsClient() + if d == nil { + return collect + } + out := collect[:0] + for _, record := range collect { + pinned, err := d.ArtifactPinned(ctx, record.SnapshotID) + if err != nil { + fm.logger.Warn("read durable artifact pin failed; not collecting", "snapshot", record.SnapshotID, "err", err) + continue + } + if pinned { + continue + } + out = append(out, record) + } + return out +} + +// the methods below are the steps of a move, in order. the durable handler +// runs each inside a journaled restate step; the in-process path calls them +// directly. each must be safe to run again. + +// ArtifactPullTimeout bounds one attempt at a pull, from +// ORCH_ARTIFACT_PULL_TIMEOUT_SECONDS. zero leaves the agent client's own +// request timeout as the bound. +func (fm *FleetManager) ArtifactPullTimeout() time.Duration { + return fm.artifactPullTimeout +} + +// ArtifactMoveCommitted returns the id a ready copy of the job is already +// recorded under on its destination, or "". a move that finds one stops +// there: a concurrent move to the same host got there first. +func (fm *FleetManager) ArtifactMoveCommitted(ctx context.Context, job ArtifactMoveJob) (string, error) { + if job.Lazy { + // a lazy copy is never recorded. + return "", nil + } + record, err := fm.getSnapshotRecord(ctx, job.LocalID) + if err != nil { + return "", nil + } + if record.HostID == job.HostID && record.State == SnapshotStateReady { + return record.SnapshotID, nil + } + return "", nil +} + +// errArtifactKindMismatch is a destination that committed something other +// than what was sent: an agent too old for live snapshots or delta merges. +// retrying cannot change it. +var errArtifactKindMismatch = errors.New("artifact committed as the wrong kind") + +// ArtifactPull has the destination fetch the bytes from the source. both +// agents' endpoints are read from the fleet here, not carried on the job. +func (fm *FleetManager) ArtifactPull(ctx context.Context, job ArtifactMoveJob) (ArtifactMoved, error) { + if fm.artifactMover == nil { + return ArtifactMoved{}, fmt.Errorf("%w: no artifact mover is configured", ErrArtifactImmovable) + } + from, err := fm.artifactEndpoint(job.SourceHostID) + if err != nil { + return ArtifactMoved{}, err + } + to, err := fm.artifactEndpoint(job.HostID) + if err != nil { + return ArtifactMoved{}, err + } + if fm.artifactPullTimeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, fm.artifactPullTimeout) + defer cancel() + } + + start := time.Now() + moved, err := fm.artifactMover.MoveArtifact(ctx, ArtifactMove{ + Digest: job.Digest, + SnapshotID: job.LocalID, + From: from, + To: to, + Files: job.Files, + Lazy: job.Lazy, + + SourceSnapshotID: job.SourceSnapshotID, + Base: job.Base, + DropBase: job.DropBase, + }) + if err == nil && job.Base != "" && moved.Kind != SnapshotKindLive { + err = fmt.Errorf("%w: host %s did not commit the merged delta as a live snapshot; its host agent cannot merge deltas", + errArtifactKindMismatch, job.HostID) + } + if err == nil && job.Base == "" && job.Kind == SnapshotKindLive && moved.Kind != SnapshotKindLive { + // the target took the rootfs and dropped the rest. nothing is + // recorded, so nothing can seed from the half it committed. + err = fmt.Errorf("%w: host %s committed only the rootfs of a live snapshot; its host agent cannot receive memory images", + errArtifactKindMismatch, job.HostID) + } + if m, ok := fm.metrics.(ArtifactMoveMetrics); ok { + m.ArtifactMoveCompleted(job.SourceHostID, job.HostID, moved.SizeBytes, time.Since(start), err) + } + if err != nil { + return ArtifactMoved{}, err + } + return moved, nil +} + +// artifactEndpoint is a registered host's agent address and token. +func (fm *FleetManager) artifactEndpoint(hostID string) (ArtifactEndpoint, error) { + fm.mu.RLock() + defer fm.mu.RUnlock() + h, ok := fm.hosts[hostID] + if !ok { + return ArtifactEndpoint{}, fmt.Errorf("%w: host %s is not registered", ErrArtifactImmovable, hostID) + } + if h.URL == "" || h.Token == "" { + return ArtifactEndpoint{}, fmt.Errorf("%w: host %s has no usable agent url or token", ErrArtifactImmovable, hostID) + } + return ArtifactEndpoint{HostID: h.ID, URL: h.URL, Token: h.Token}, nil +} + +// PermanentArtifactError reports whether a failed pull cannot succeed on a +// retry: the artifact cannot move at all, the destination committed the wrong +// kind, or an agent refused the request with a 4xx other than a timeout or +// rate limit. +func PermanentArtifactError(err error) bool { + if errors.Is(err, ErrArtifactImmovable) || errors.Is(err, errArtifactKindMismatch) { + return true + } + var statusErr *HTTPStatusError + if errors.As(err, &statusErr) { + code := statusErr.Code + return code >= 400 && code < 500 && code != http.StatusRequestTimeout && code != http.StatusTooManyRequests + } + return false +} + +// ArtifactRecord records the copy the destination committed, and returns its +// id. a lazy copy is not recorded: its agent deletes it once the vm resumed. +func (fm *FleetManager) ArtifactRecord(ctx context.Context, job ArtifactMoveJob, moved ArtifactMoved) (string, error) { + committedID := moved.SnapshotID + if committedID == "" { + committedID = job.LocalID + } + if job.Lazy { + // the target holds no memory image, so it can never be the source of + // the next move or a seed for anything else, and its agent deletes the + // seed once the vm has resumed. recording it would say otherwise. + return committedID, nil + } + if job.Base != "" { + return fm.recordMergedDelta(ctx, job, moved, committedID) + } + size := moved.SizeBytes + if size <= 0 { + size = job.SizeBytes + } + + now := time.Now() + replicaMetadata := map[string]string{ + "source_snapshot_id": job.SnapshotID, + "source_host_id": job.SourceHostID, + } + if len(job.Files) > 0 { + // the copy can be a source for the next move, so it carries the digests. + replicaMetadata = withLiveFiles(replicaMetadata, job.Files) + } + metadata, _ := marshalSnapshotMetadata("", replicaMetadata) + replica := SnapshotRecord{ + SnapshotID: committedID, + // No VM: this copy was not checkpointed from anything running here, it + // arrived over the wire. Leaving VMID empty is what marks it as a + // free-standing artifact rather than a vm's checkpoint. + VMID: "", + HostID: job.HostID, + TenantID: job.TenantID, + // ParentSnapshotID is deliberately NOT the source. Lineage means "this + // snapshot descends from that one", and deleting a snapshot with + // children is refused; pointing at the source would make every origin + // artifact undeletable the moment anything pulled a copy of it. + Mode: SnapshotModeBuild, + LayerKey: job.LayerKey, + Arch: job.Arch, + Digest: job.Digest, + Kind: job.Kind, + State: SnapshotStateReady, + SizeBytes: size, + Metadata: metadata, + CreatedAt: now, + UpdatedAt: now, + } + if err := fm.upsertSnapshotRecord(ctx, replica); err != nil { + // The bytes are on the host but nothing records them. Failing is the + // consistent choice: the create that triggered this would fail at its + // own next persist anyway, and the id is derived from (digest, host) so + // the next attempt re-commits over the same directory rather than + // leaving one orphan per try. + return "", fmt.Errorf("persist pulled artifact %s on host %s: %w", committedID, job.HostID, err) + } + + fm.appendEventBackground("host", job.HostID, "artifact.pulled", map[string]any{ + "digest": job.Digest, + "snapshot_id": committedID, + "source_host_id": job.SourceHostID, + "source_snapshot_id": job.SnapshotID, + "size_bytes": size, + }) + return committedID, nil +} + +// recordMergedDelta records the complete live snapshot a delta merge left on +// the destination, and forgets the base the merge dropped. +func (fm *FleetManager) recordMergedDelta(ctx context.Context, job ArtifactMoveJob, moved ArtifactMoved, committedID string) (string, error) { + now := time.Now() + replicaMetadata := map[string]string{ + "source_snapshot_id": job.SnapshotID, + "source_host_id": job.SourceHostID, + "merged_onto": job.Base, + } + if completeLiveFiles(moved.Files) { + replicaMetadata = withLiveFiles(replicaMetadata, moved.Files) + } + metadata, _ := marshalSnapshotMetadata("", replicaMetadata) + replica := SnapshotRecord{ + SnapshotID: committedID, + HostID: job.HostID, + TenantID: job.TenantID, + // build mode keeps retention gc away from it: the code that made it + // (a migrate or the checkpoint loop) is the code that removes it. + Mode: SnapshotModeBuild, + Arch: job.Arch, + Digest: job.Digest, + Kind: SnapshotKindLive, + State: SnapshotStateReady, + SizeBytes: moved.SizeBytes, + Metadata: metadata, + CreatedAt: now, + UpdatedAt: now, + } + if err := fm.upsertSnapshotRecord(ctx, replica); err != nil { + return "", fmt.Errorf("persist merged delta %s on host %s: %w", committedID, job.HostID, err) + } + if job.DropBase { + fm.forgetSnapshotRecord(ctx, job.Base) + } + return committedID, nil +} + +// ArtifactDiscard deletes what a failed move may have left on the +// destination. a copy that did get recorded is kept: the record step can +// commit on an attempt whose answer was lost. +func (fm *FleetManager) ArtifactDiscard(ctx context.Context, job ArtifactMoveJob) error { + if record, err := fm.getSnapshotRecord(ctx, job.LocalID); err == nil && record.HostID == job.HostID { + return nil + } + if err := fm.deleteFreeArtifact(ctx, job.HostID, job.LocalID); err != nil && !IsNotFound(err) { + return fmt.Errorf("delete partial copy %s on host %s: %w", job.LocalID, job.HostID, err) + } + fm.logger.InfoContext(ctx, "deleted partial artifact copy after a failed move", "snapshot", job.LocalID, "host", job.HostID) + return nil +} diff --git a/internal/orchestrator/artifacts_durable_test.go b/internal/orchestrator/artifacts_durable_test.go new file mode 100644 index 00000000..65f84d62 --- /dev/null +++ b/internal/orchestrator/artifacts_durable_test.go @@ -0,0 +1,303 @@ +package orchestrator + +import ( + "context" + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" +) + +// fakeDurableArtifacts stands in for the restate client: it records the jobs +// it was sent and the pins it holds. +type fakeDurableArtifacts struct { + mu sync.Mutex + jobs []ArtifactMoveJob + moveErr error + pinErr error + pins map[string]map[string]time.Duration +} + +func newFakeDurableArtifacts() *fakeDurableArtifacts { + return &fakeDurableArtifacts{pins: map[string]map[string]time.Duration{}} +} + +func (f *fakeDurableArtifacts) MoveArtifact(_ context.Context, job ArtifactMoveJob) (string, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.jobs = append(f.jobs, job) + if f.moveErr != nil { + return "", f.moveErr + } + return job.LocalID, nil +} + +func (f *fakeDurableArtifacts) PinArtifacts(_ context.Context, holder string, ttl time.Duration, ids ...string) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.pinErr != nil { + return f.pinErr + } + for _, id := range ids { + if f.pins[id] == nil { + f.pins[id] = map[string]time.Duration{} + } + f.pins[id][holder] = ttl + } + return nil +} + +func (f *fakeDurableArtifacts) ReleaseArtifacts(_ context.Context, holder string, ids ...string) error { + f.mu.Lock() + defer f.mu.Unlock() + for _, id := range ids { + delete(f.pins[id], holder) + } + return nil +} + +func (f *fakeDurableArtifacts) ArtifactPinned(_ context.Context, id string) (bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.pinErr != nil { + return false, f.pinErr + } + return len(f.pins[id]) > 0, nil +} + +// deletingProvider is a stub provider that can delete free-standing copies. +type deletingProvider struct { + *stubLoadProvider + mu sync.Mutex + deleted []string +} + +func (p *deletingProvider) DeleteArtifact(_ context.Context, snapshotID string) error { + p.mu.Lock() + defer p.mu.Unlock() + p.deleted = append(p.deleted, snapshotID) + return nil +} + +func twoHostFleet(t *testing.T, mover ArtifactMover) (*FleetManager, *deletingProvider) { + t.Helper() + stub := &deletingProvider{stubLoadProvider: newStubProvider()} + fm := NewFleetManager(FleetConfig{Provider: stub, Prefix: "fuse-", ArtifactMover: mover}) + for _, id := range []string{"h-a", "h-b"} { + if err := fm.RegisterHost(context.Background(), artifactHost(id, 8), stub); err != nil { + t.Fatal(err) + } + } + return fm, stub +} + +// TestEnsureArtifactOnHost_durable: with restate configured the move goes to +// restate, carrying host ids and no agent token, and the in-process mover is +// never called. +func TestEnsureArtifactOnHost_durable(t *testing.T) { + ctx := context.Background() + mover := &recordingMover{} + fm, _ := twoHostFleet(t, mover) + d := newFakeDurableArtifacts() + fm.SetDurableArtifacts(d) + record := seedArtifactRecord(t, fm, "art-1", "tenant-1", "h-a", "d1") + + got, err := fm.ensureArtifactOnHost(ctx, record, "h-b", false) + if err != nil { + t.Fatalf("ensure: %v", err) + } + if want := replicaSnapshotID("d1", "h-b"); got != want { + t.Fatalf("committed %q, want %q", got, want) + } + if n := len(mover.calls()); n != 0 { + t.Fatalf("in-process mover called %d times, want 0", n) + } + if len(d.jobs) != 1 { + t.Fatalf("%d durable moves, want 1", len(d.jobs)) + } + job := d.jobs[0] + if job.SourceHostID != "h-a" || job.HostID != "h-b" || job.SnapshotID != "art-1" || job.TenantID != "tenant-1" { + t.Fatalf("job %+v names the wrong ends", job) + } + if strings.Contains(fmt.Sprintf("%+v", job), "token-") { + t.Fatalf("job %+v carries an agent token into the journal", job) + } + + d.moveErr = errors.New("restate said no") + _, err = fm.ensureArtifactOnHost(ctx, seedArtifactRecord(t, fm, "art-2", "tenant-1", "h-a", "d2"), "h-b", false) + if !errors.Is(err, ErrArtifactTransferFailed) { + t.Fatalf("failed durable move: %v, want ErrArtifactTransferFailed", err) + } +} + +// TestArtifactSteps: pull resolves endpoints from the fleet and record writes +// the copy, so the durable handler and the in-process path record the same. +func TestArtifactSteps(t *testing.T) { + ctx := context.Background() + mover := &recordingMover{} + fm, _ := twoHostFleet(t, mover) + job := ArtifactMoveJob{ + SnapshotID: "art-1", Digest: "d1", SourceHostID: "h-a", SourceSnapshotID: "art-1", + HostID: "h-b", LocalID: "art-copy", TenantID: "tenant-1", LayerKey: "layer-1", SizeBytes: 7, + } + if got, err := fm.ArtifactMoveCommitted(ctx, job); err != nil || got != "" { + t.Fatalf("committed before any move = %q, %v", got, err) + } + moved, err := fm.ArtifactPull(ctx, job) + if err != nil { + t.Fatalf("pull: %v", err) + } + calls := mover.calls() + if len(calls) != 1 || calls[0].From.Token != "token-h-a" || calls[0].To.Token != "token-h-b" { + t.Fatalf("mover calls %+v, want one from h-a to h-b with their tokens", calls) + } + id, err := fm.ArtifactRecord(ctx, job, moved) + if err != nil || id != "art-copy" { + t.Fatalf("record = %q, %v", id, err) + } + rec, err := fm.GetSnapshotByID(ctx, "art-copy") + if err != nil || rec.HostID != "h-b" || rec.LayerKey != "layer-1" || rec.Digest != "d1" { + t.Fatalf("recorded copy %+v, %v", rec, err) + } + // a second move into the same copy stops at the check. + if got, err := fm.ArtifactMoveCommitted(ctx, job); err != nil || got != "art-copy" { + t.Fatalf("committed after the move = %q, %v", got, err) + } + + gone := job + gone.SourceHostID = "h-gone" + if _, err := fm.ArtifactPull(ctx, gone); !PermanentArtifactError(err) { + t.Fatalf("pull from an unregistered host: %v, want a permanent error", err) + } +} + +// TestArtifactDiscard deletes an unrecorded copy and keeps a recorded one. +func TestArtifactDiscard(t *testing.T) { + ctx := context.Background() + fm, stub := twoHostFleet(t, &recordingMover{}) + job := ArtifactMoveJob{HostID: "h-b", LocalID: "art-partial"} + if err := fm.ArtifactDiscard(ctx, job); err != nil { + t.Fatal(err) + } + seedArtifactRecord(t, fm, "art-kept", "tenant-1", "h-b", "d1") + if err := fm.ArtifactDiscard(ctx, ArtifactMoveJob{HostID: "h-b", LocalID: "art-kept"}); err != nil { + t.Fatal(err) + } + if fmt.Sprint(stub.deleted) != "[art-partial]" { + t.Fatalf("deleted %v, want only the unrecorded copy", stub.deleted) + } +} + +func TestPermanentArtifactError(t *testing.T) { + cases := []struct { + err error + want bool + }{ + {fmt.Errorf("x: %w", ErrArtifactImmovable), true}, + {fmt.Errorf("x: %w", errArtifactKindMismatch), true}, + {fmt.Errorf("pull: %w", &HTTPStatusError{Code: 409}), true}, + {fmt.Errorf("pull: %w", &HTTPStatusError{Code: 429}), false}, + {fmt.Errorf("pull: %w", &HTTPStatusError{Code: 502}), false}, + {errors.New("connection reset"), false}, + {context.DeadlineExceeded, false}, + } + for _, c := range cases { + if got := PermanentArtifactError(c.err); got != c.want { + t.Errorf("PermanentArtifactError(%v) = %v, want %v", c.err, got, c.want) + } + } +} + +// TestReconcileArtifacts_durablePins: gc leaves an artifact a durable pin +// holds, and one whose pin cannot be read. +func TestReconcileArtifacts_durablePins(t *testing.T) { + cases := []struct { + name string + arrange func(d *fakeDurableArtifacts) + wantAlive bool + }{ + {"unpinned is collected", func(*fakeDurableArtifacts) {}, false}, + {"pinned survives", func(d *fakeDurableArtifacts) { + _ = d.PinArtifacts(context.Background(), "move:inv", time.Hour, "art-1") + }, true}, + {"unreadable pin survives", func(d *fakeDurableArtifacts) { d.pinErr = errors.New("restate down") }, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + fm := NewFleetManager(FleetConfig{Provider: newStubProvider(), Prefix: "fuse-", ArtifactIdleTTL: time.Minute}) + d := newFakeDurableArtifacts() + fm.SetDurableArtifacts(d) + record := seedArtifactRecord(t, fm, "art-1", "tenant-1", "h-a", "d1") + stale := time.Now().Add(-24 * time.Hour) + record.CreatedAt, record.UpdatedAt = stale, stale + if err := fm.upsertSnapshotRecord(ctx, record); err != nil { + t.Fatal(err) + } + tc.arrange(d) + fm.reconcileArtifacts(ctx) + _, err := fm.GetSnapshotByID(ctx, "art-1") + if alive := err == nil; alive != tc.wantAlive { + t.Fatalf("alive = %v, want %v", alive, tc.wantAlive) + } + }) + } +} + +// TestHoldArtifact: on restate the pin is durable and released by the caller; +// a pin restate refuses is held in memory instead. +func TestHoldArtifact(t *testing.T) { + ctx := context.Background() + fm := NewFleetManager(FleetConfig{Provider: newStubProvider(), Prefix: "fuse-"}) + d := newFakeDurableArtifacts() + fm.SetDurableArtifacts(d) + + release := fm.holdArtifact(ctx, "lazy:vm-1", lazyMemoryPinTTL, "seed-1") + if p, _ := d.ArtifactPinned(ctx, "seed-1"); !p || len(fm.artifactPulls) != 0 { + t.Fatalf("durable pin %v, in-memory pins %v", p, fm.artifactPulls) + } + release() + if p, _ := d.ArtifactPinned(ctx, "seed-1"); p { + t.Fatal("release left the durable pin") + } + + d.pinErr = errors.New("restate down") + release = fm.holdArtifact(ctx, "lazy:vm-2", lazyMemoryPinTTL, "seed-2") + if fm.artifactPulls["seed-2"] != 1 { + t.Fatalf("in-memory pins %v, want seed-2 held", fm.artifactPulls) + } + release() + if len(fm.artifactPulls) != 0 { + t.Fatalf("in-memory pins %v after release", fm.artifactPulls) + } +} + +// moveMetrics records ArtifactMoveCompleted. +type moveMetrics struct { + got []string +} + +func (*moveMetrics) ReconcileCompleted(ReconcileSummary) {} + +func (m *moveMetrics) ArtifactMoveCompleted(src, dst string, bytes int64, _ time.Duration, err error) { + m.got = append(m.got, fmt.Sprintf("%s->%s %d %v", src, dst, bytes, err != nil)) +} + +func TestArtifactPull_metrics(t *testing.T) { + ctx := context.Background() + mover := &recordingMover{} + fm, _ := twoHostFleet(t, mover) + m := &moveMetrics{} + fm.metrics = m + job := ArtifactMoveJob{SnapshotID: "art-1", Digest: "d1", SourceHostID: "h-a", HostID: "h-b", LocalID: "c"} + if _, err := fm.ArtifactPull(ctx, job); err != nil { + t.Fatal(err) + } + mover.err = errors.New("peer reset") + _, _ = fm.ArtifactPull(ctx, job) + if fmt.Sprint(m.got) != "[h-a->h-b 4096 false h-a->h-b 0 true]" { + t.Fatalf("metrics %v", m.got) + } +} diff --git a/internal/orchestrator/delta.go b/internal/orchestrator/delta.go index 33338d3c..503c6671 100644 --- a/internal/orchestrator/delta.go +++ b/internal/orchestrator/delta.go @@ -3,7 +3,6 @@ package orchestrator import ( "context" "fmt" - "time" ) // deleteFreeArtifact removes a free-standing copy (a pulled artifact or a @@ -59,79 +58,32 @@ func (fm *FleetManager) moveDeltaToHost(ctx context.Context, child SnapshotRecor return "", fmt.Errorf("%w: diff snapshot %s has no recorded digests for its delta files", ErrArtifactImmovable, child.SnapshotID) } holders := map[string]ArtifactHolder{child.HostID: {HostID: child.HostID, SnapshotID: child.SnapshotID}} - source, target, err := fm.artifactEndpoints(child, holders, hostID) + source, _, err := fm.artifactEndpoints(child, holders, hostID) if err != nil { return "", err } - localID := liveReplicaSnapshotID(child.SnapshotID, hostID) - fm.beginArtifactPull(child.SnapshotID, base, localID) - defer fm.endArtifactPull(child.SnapshotID, base, localID) - - pullCtx := ctx - if fm.artifactPullTimeout > 0 { - var cancel context.CancelFunc - pullCtx, cancel = context.WithTimeout(ctx, fm.artifactPullTimeout) - defer cancel() - } - moved, err := fm.artifactMover.MoveArtifact(pullCtx, ArtifactMove{ + job := ArtifactMoveJob{ + SnapshotID: child.SnapshotID, Digest: child.Digest, - SnapshotID: localID, - From: source.endpoint, - To: target, - Files: files, + SourceHostID: source.endpoint.HostID, SourceSnapshotID: child.SnapshotID, + HostID: hostID, + LocalID: liveReplicaSnapshotID(child.SnapshotID, hostID), + Files: files, Base: base, DropBase: dropBase, - }) - if err == nil && moved.Kind != SnapshotKindLive { - err = fmt.Errorf("host %s did not commit the merged delta as a live snapshot; its host agent cannot merge deltas", hostID) + TenantID: child.TenantID, + Arch: child.Arch, + Kind: SnapshotKindLive, } - if err != nil { + return fm.runArtifactMove(ctx, job, func(err error) error { if ctxErr := ctx.Err(); ctxErr != nil { - return "", fmt.Errorf("move delta %s to host %s: %w", child.SnapshotID, hostID, ctxErr) + return fmt.Errorf("move delta %s to host %s: %w", child.SnapshotID, hostID, ctxErr) } - return "", fmt.Errorf("%w: move delta %s from host %s onto %s on host %s: %w", + return fmt.Errorf("%w: move delta %s from host %s onto %s on host %s: %w", ErrArtifactTransferFailed, child.SnapshotID, source.endpoint.HostID, base, hostID, err) - } - - committedID := moved.SnapshotID - if committedID == "" { - committedID = localID - } - now := time.Now() - replicaMetadata := map[string]string{ - "source_snapshot_id": child.SnapshotID, - "source_host_id": source.endpoint.HostID, - "merged_onto": base, - } - if completeLiveFiles(moved.Files) { - replicaMetadata = withLiveFiles(replicaMetadata, moved.Files) - } - metadata, _ := marshalSnapshotMetadata("", replicaMetadata) - replica := SnapshotRecord{ - SnapshotID: committedID, - HostID: hostID, - TenantID: child.TenantID, - // build mode keeps retention gc away from it: the code that made it - // (a migrate or the checkpoint loop) is the code that removes it. - Mode: SnapshotModeBuild, - Arch: child.Arch, - Digest: child.Digest, - Kind: SnapshotKindLive, - State: SnapshotStateReady, - SizeBytes: moved.SizeBytes, - Metadata: metadata, - CreatedAt: now, - UpdatedAt: now, - } - if err := fm.upsertSnapshotRecord(ctx, replica); err != nil { - return "", fmt.Errorf("persist merged delta %s on host %s: %w", committedID, hostID, err) - } - if dropBase { - fm.forgetSnapshotRecord(ctx, base) - } - return committedID, nil + }) } // completeLiveFiles reports whether files names a digest for every file a diff --git a/internal/orchestrator/fleet.go b/internal/orchestrator/fleet.go index 50b11b08..683a955a 100644 --- a/internal/orchestrator/fleet.go +++ b/internal/orchestrator/fleet.go @@ -621,6 +621,10 @@ type FleetManager struct { destroySendSlots chan struct{} destroySends sync.WaitGroup + // durableArtifacts runs artifact moves and holds pull pins on restate; + // nil in mode off. see artifacts_durable.go. + durableArtifacts DurableArtifacts + cancel context.CancelFunc done chan struct{} } diff --git a/internal/orchestrator/migrate.go b/internal/orchestrator/migrate.go index a707d137..1eabe1dc 100644 --- a/internal/orchestrator/migrate.go +++ b/internal/orchestrator/migrate.go @@ -421,9 +421,11 @@ func (fm *FleetManager) migrateVM(ctx context.Context, vmID string, opts Migrate if opts.Lazy { // the seed on the source is now the only copy of every page the new // vm has not touched yet, so it is held until the target has them all. - // taken before the source vm goes, and released by the watcher. - fm.beginArtifactPull(seed.SnapshotID) - go fm.watchLazyMemory(targetProvider, newVMID, seed.SnapshotID) + // taken before the source vm goes, and released by the watcher. on + // restate the pin outlives a restart that loses the watcher, and + // lapses once no chunk could still arrive. + release := fm.holdArtifact(ctx, "lazy:"+newVMID, lazyMemoryPinTTL, seed.SnapshotID) + go fm.watchLazyMemory(targetProvider, newVMID, seed.SnapshotID, release) } // from here the target is the guest; the source is only torn down. @@ -453,6 +455,10 @@ func (fm *FleetManager) migrateVM(ctx context.Context, vmID string, opts Migrate // target's grant is dead and no further chunk can arrive. const lazyMemoryWatchTimeout = 30 * time.Minute +// lazyMemoryPinTTL is how long a durable lazy-memory pin lasts when no +// watcher releases it: a little past the grant. +const lazyMemoryPinTTL = lazyMemoryWatchTimeout + 5*time.Minute + // lazyMemoryPollInterval is how often the target is asked how far along it is. var lazyMemoryPollInterval = 2 * time.Second @@ -460,8 +466,9 @@ var lazyMemoryPollInterval = 2 * time.Second // target, then releases the seed it was paging from. the outcome is an event // either way: a failed transfer leaves a guest that will hang on its next // missing page, and the only thing the orchestrator can do is say so. -func (fm *FleetManager) watchLazyMemory(provider Provider, vmID, seedID string) { - defer fm.endArtifactPull(seedID) +// release drops the pin on the seed. +func (fm *FleetManager) watchLazyMemory(provider Provider, vmID, seedID string, release func()) { + defer release() w, ok := provider.(MemoryWatcher) if !ok { return From 418f21c29d016ca89572bef9291062d7fa8b2404 Mon Sep 17 00:00:00 2001 From: andrewn6 <59238070+andrewn6@users.noreply.github.com> Date: Sat, 14 Nov 2026 17:45:38 -0500 Subject: [PATCH 2/6] feat: run artifact moves and pull pins as restate Artifacts and ArtifactPin objects --- internal/durable/artifacts.go | 236 ++++++++++++++++++++ internal/durable/artifacts_test.go | 333 +++++++++++++++++++++++++++++ internal/durable/client.go | 24 ++- internal/durable/destroy_test.go | 20 +- internal/durable/server.go | 37 +++- internal/durable/server_test.go | 19 +- orchestrator/main.go | 4 +- 7 files changed, 656 insertions(+), 17 deletions(-) create mode 100644 internal/durable/artifacts.go create mode 100644 internal/durable/artifacts_test.go diff --git a/internal/durable/artifacts.go b/internal/durable/artifacts.go new file mode 100644 index 00000000..3cb07dd2 --- /dev/null +++ b/internal/durable/artifacts.go @@ -0,0 +1,236 @@ +package durable + +import ( + "context" + "time" + + restate "github.com/restatedev/sdk-go" + "github.com/restatedev/sdk-go/ingress" + + "github.com/folsomintel/fuse/internal/orchestrator" +) + +// opArtifactMove is the op label on a move's step spans and metrics. +const opArtifactMove = "artifact_move" + +// movePinTTL is how long a move's pins last if the move never releases them, +// as when its invocation is killed. a move that finishes, fails or is +// cancelled releases them itself; this only stops a dead one holding an +// artifact forever. +const movePinTTL = 24 * time.Hour + +// defaultPullAttemptBound is how long one pull attempt can take with no +// ORCH_ARTIFACT_PULL_TIMEOUT_SECONDS: the agent client's 10 minute request +// timeout, three times over for its keyed retries, plus their backoff. +const defaultPullAttemptBound = 35 * time.Minute + +// Artifacts is the part of the fleet the move handler drives, one method per +// step. *FleetManager satisfies it. +type Artifacts interface { + ArtifactMoveCommitted(ctx context.Context, job orchestrator.ArtifactMoveJob) (string, error) + ArtifactPull(ctx context.Context, job orchestrator.ArtifactMoveJob) (orchestrator.ArtifactMoved, error) + ArtifactRecord(ctx context.Context, job orchestrator.ArtifactMoveJob, moved orchestrator.ArtifactMoved) (string, error) + ArtifactDiscard(ctx context.Context, job orchestrator.ArtifactMoveJob) error + ArtifactPullTimeout() time.Duration +} + +// moveAbortTimeout is how long restate lets a move attempt run without +// journal progress before it aborts and retries it. a pull is one long step, +// so it has to outlast one pull attempt. +func moveAbortTimeout(pullTimeout time.Duration) time.Duration { + bound := defaultPullAttemptBound + if pullTimeout > 0 { + bound = pullTimeout + } + return bound + 5*time.Minute +} + +// move is Artifacts/{local_id}.Move. keying on the destination's snapshot id +// runs one move into a given copy at a time; a second one finds the first's +// copy recorded and returns it. the pins on both ends live in ArtifactPin +// state for the whole move, so gc leaves them alone across a restart. a pull +// or record that fails for good deletes what reached the destination, so no +// copy is left there that nothing records. +func move(ctx restate.ObjectContext, fleet Artifacts, policy RetryPolicy, job orchestrator.ArtifactMoveJob) (string, error) { + if key := restate.Key(ctx); job.LocalID != key { + return "", restate.TerminalErrorf("move into %s sent to Artifacts/%s", job.LocalID, key) + } + + committed, err := Run(ctx, opArtifactMove, "check", func(c context.Context) (string, error) { + return fleet.ArtifactMoveCommitted(c, job) + }) + if err != nil || committed != "" { + return committed, err + } + + holder := "move:" + ctx.Request().ID + ids := job.PinIDs() + for _, id := range ids { + if _, err := restate.Object[restate.Void](ctx, ArtifactPinObject, id, "Pin"). + Request(PinRequest{Holder: holder, TTL: movePinTTL}); err != nil { + return "", err + } + } + + retry := []restate.RunOption{ + restate.WithInitialRetryInterval(policy.InitialInterval), + restate.WithMaxRetryInterval(policy.MaxInterval), + restate.WithMaxRetryAttempts(uint(policy.MaxAttempts)), + } + moved, err := Run(ctx, opArtifactMove, "pull", func(c context.Context) (orchestrator.ArtifactMoved, error) { + moved, err := fleet.ArtifactPull(c, job) + if err != nil && orchestrator.PermanentArtifactError(err) { + return moved, restate.ToTerminalError(err) + } + return moved, err + }, retry...) + var id string + if err == nil { + id, err = Run(ctx, opArtifactMove, "record", func(c context.Context) (string, error) { + return fleet.ArtifactRecord(c, job, moved) + }, retry...) + } + if err != nil { + if !restate.IsTerminalError(err) { + // suspension: restate resumes the move here. + return "", err + } + if _, derr := Run(ctx, opArtifactMove, "discard", func(c context.Context) (restate.Void, error) { + return restate.Void{}, fleet.ArtifactDiscard(c, job) + }, retry...); derr != nil && !restate.IsTerminalError(derr) { + return "", derr + } + releasePins(ctx, holder, ids) + return "", err + } + releasePins(ctx, holder, ids) + return id, nil +} + +func releasePins(ctx restate.Context, holder string, ids []string) { + for _, id := range ids { + restate.ObjectSend(ctx, ArtifactPinObject, id, "Release").Send(holder) + } +} + +// PinRequest holds an ArtifactPin for Holder, until released or TTL passes. +type PinRequest struct { + Holder string `json:"holder"` + TTL time.Duration `json:"ttl"` +} + +// pinExpiry is the delayed send that lapses one pin. Deadline tells a lapse +// apart from one a later Pin by the same holder has already extended. +type pinExpiry struct { + Holder string `json:"holder"` + Deadline int64 `json:"deadline"` +} + +// pinHoldersKey is the ArtifactPin state: holder to deadline, unix ms. +const pinHoldersKey = "holders" + +// pin is ArtifactPin/{snapshot_id}.Pin. the lapse is a delayed send rather +// than a sleeping invocation, so a pin held for hours pins nothing to this +// deployment. +func pin(ctx restate.ObjectContext, req PinRequest) error { + if req.Holder == "" || req.TTL <= 0 { + return restate.TerminalErrorf("pin needs a holder and a positive ttl") + } + now, err := restate.Run(ctx, func(restate.RunContext) (int64, error) { + return time.Now().UnixMilli(), nil + }, restate.WithName("now")) + if err != nil { + return err + } + holders, err := restate.Get[map[string]int64](ctx, pinHoldersKey) + if err != nil { + return err + } + if holders == nil { + holders = map[string]int64{} + } + deadline := now + req.TTL.Milliseconds() + holders[req.Holder] = deadline + restate.Set(ctx, pinHoldersKey, holders) + restate.ObjectSend(ctx, ArtifactPinObject, restate.Key(ctx), "Expire"). + Send(pinExpiry{Holder: req.Holder, Deadline: deadline}, restate.WithDelay(req.TTL)) + return nil +} + +// release is ArtifactPin/{snapshot_id}.Release. +func release(ctx restate.ObjectContext, holder string) error { + return dropHolder(ctx, holder, 0) +} + +// expire is ArtifactPin/{snapshot_id}.Expire, the pin's timer. +func expire(ctx restate.ObjectContext, e pinExpiry) error { + return dropHolder(ctx, e.Holder, e.Deadline) +} + +// dropHolder removes holder, or with a deadline only the pin that deadline +// belongs to. the last holder out clears the object. +func dropHolder(ctx restate.ObjectContext, holder string, deadline int64) error { + holders, err := restate.Get[map[string]int64](ctx, pinHoldersKey) + if err != nil { + return err + } + got, ok := holders[holder] + if !ok || (deadline != 0 && got != deadline) { + return nil + } + delete(holders, holder) + if len(holders) == 0 { + restate.ClearAll(ctx) + return nil + } + restate.Set(ctx, pinHoldersKey, holders) + return nil +} + +// pinned is ArtifactPin/{snapshot_id}.Pinned. +func pinned(ctx restate.ObjectSharedContext) (bool, error) { + holders, err := restate.Get[map[string]int64](ctx, pinHoldersKey) + if err != nil { + return false, err + } + return len(holders) > 0, nil +} + +// MoveArtifact runs Artifacts/{job.LocalID}.Move and waits for it. a move +// takes as long as its bytes do, so it goes through the client with no +// request timeout of its own; ctx bounds the wait, and the move keeps running +// if ctx ends first. +func (c *Client) MoveArtifact(ctx context.Context, job orchestrator.ArtifactMoveJob) (string, error) { + return ingress.Object[orchestrator.ArtifactMoveJob, string](c.longIngress, ArtifactsObject, job.LocalID, "Move"). + Request(ctx, job) +} + +// PinArtifacts pins each id for holder, for at most ttl. +func (c *Client) PinArtifacts(ctx context.Context, holder string, ttl time.Duration, ids ...string) error { + for _, id := range ids { + if _, err := ingress.Object[PinRequest, restate.Void](c.ingress, ArtifactPinObject, id, "Pin"). + Request(ctx, PinRequest{Holder: holder, TTL: ttl}); err != nil { + return err + } + } + return nil +} + +// ReleaseArtifacts drops holder's pin on each id. +func (c *Client) ReleaseArtifacts(ctx context.Context, holder string, ids ...string) error { + for _, id := range ids { + if _, err := ingress.Object[string, restate.Void](c.ingress, ArtifactPinObject, id, "Release"). + Request(ctx, holder); err != nil { + return err + } + } + return nil +} + +// ArtifactPinned reports whether any holder pins id. +func (c *Client) ArtifactPinned(ctx context.Context, id string) (bool, error) { + return ingress.Object[restate.Void, bool](c.ingress, ArtifactPinObject, id, "Pinned"). + Request(ctx, restate.Void{}) +} + +var _ orchestrator.DurableArtifacts = (*Client)(nil) diff --git a/internal/durable/artifacts_test.go b/internal/durable/artifacts_test.go new file mode 100644 index 00000000..47e87681 --- /dev/null +++ b/internal/durable/artifacts_test.go @@ -0,0 +1,333 @@ +package durable + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "strings" + "sync" + "testing" + "time" + + "github.com/folsomintel/fuse/internal/orchestrator" +) + +// moveFleet scripts the move steps. pull returns pullErrs in turn, then +// succeeds; block, when set, holds a pull until it is closed. +type moveFleet struct { + fakeFleet + committed string + + mu sync.Mutex + calls []string + pullErrs []error + block chan struct{} + pulling chan struct{} +} + +func (f *moveFleet) record(step string) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, step) +} + +func (f *moveFleet) steps() string { + f.mu.Lock() + defer f.mu.Unlock() + return strings.Join(f.calls, ",") +} + +func (f *moveFleet) ArtifactMoveCommitted(context.Context, orchestrator.ArtifactMoveJob) (string, error) { + f.record("check") + return f.committed, nil +} + +func (f *moveFleet) ArtifactPull(ctx context.Context, job orchestrator.ArtifactMoveJob) (orchestrator.ArtifactMoved, error) { + f.record("pull") + if f.pulling != nil { + select { + case f.pulling <- struct{}{}: + default: + } + } + if f.block != nil { + select { + case <-f.block: + case <-ctx.Done(): + return orchestrator.ArtifactMoved{}, ctx.Err() + } + } + f.mu.Lock() + defer f.mu.Unlock() + if len(f.pullErrs) > 0 { + err := f.pullErrs[0] + f.pullErrs = f.pullErrs[1:] + return orchestrator.ArtifactMoved{}, err + } + return orchestrator.ArtifactMoved{SnapshotID: job.LocalID, SizeBytes: 42}, nil +} + +func (f *moveFleet) ArtifactRecord(_ context.Context, job orchestrator.ArtifactMoveJob, moved orchestrator.ArtifactMoved) (string, error) { + f.record("record") + return moved.SnapshotID, nil +} + +func (f *moveFleet) ArtifactDiscard(context.Context, orchestrator.ArtifactMoveJob) error { + f.record("discard") + return nil +} + +func testJob(name string) orchestrator.ArtifactMoveJob { + id := fmt.Sprintf("%s-%d", name, time.Now().UnixNano()) + return orchestrator.ArtifactMoveJob{ + SnapshotID: id + "-src", SourceSnapshotID: id + "-src", SourceHostID: "h1", + HostID: "h2", LocalID: id + "-dst", Digest: "d1", + } +} + +// pinState reads Pinned for each id. +func pinState(t *testing.T, ctx context.Context, client *Client, ids []string) []bool { + t.Helper() + out := make([]bool, len(ids)) + for i, id := range ids { + p, err := client.ArtifactPinned(ctx, id) + if err != nil { + t.Fatalf("pinned %s: %v", id, err) + } + out[i] = p + } + return out +} + +// waitUnpinned polls until no id is pinned: releases are one-way sends. +func waitUnpinned(t *testing.T, ctx context.Context, client *Client, ids []string) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for { + held := false + for _, p := range pinState(t, ctx, client, ids) { + held = held || p + } + if !held { + return + } + if time.Now().After(deadline) { + t.Fatalf("pins on %v never released", ids) + } + time.Sleep(50 * time.Millisecond) + } +} + +// TestMoveThroughRestate: a pull that fails twice is retried inside the pull +// step, the copy is recorded once, and the pins are held while it runs and +// released after. +func TestMoveThroughRestate(t *testing.T) { + fleet := &moveFleet{ + pullErrs: []error{errors.New("peer reset"), errors.New("peer reset")}, + block: make(chan struct{}), + pulling: make(chan struct{}, 1), + } + client := serveFleet(t, fleet, fastPolicy) + job := testJob("art-move") + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + done := make(chan error, 1) + var got string + go func() { + var err error + got, err = client.MoveArtifact(ctx, job) + done <- err + }() + + <-fleet.pulling + for i, p := range pinState(t, ctx, client, job.PinIDs()) { + if !p { + t.Fatalf("%s not pinned during the pull", job.PinIDs()[i]) + } + } + close(fleet.block) + + if err := <-done; err != nil { + t.Fatalf("move: %v", err) + } + if got != job.LocalID { + t.Fatalf("move returned %q, want %q", got, job.LocalID) + } + if got, want := fleet.steps(), "check,pull,pull,pull,record"; got != want { + t.Fatalf("steps %s, want %s", got, want) + } + waitUnpinned(t, ctx, client, job.PinIDs()) +} + +// TestMoveDiscardsAfterPermanentFailure: an agent that refuses the pull is not +// retried, the partial copy is discarded and the pins are released. +func TestMoveDiscardsAfterPermanentFailure(t *testing.T) { + fleet := &moveFleet{pullErrs: []error{&orchestrator.HTTPStatusError{Code: 409, Body: "digest mismatch"}}} + client := serveFleet(t, fleet, fastPolicy) + job := testJob("art-refused") + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + _, err := client.MoveArtifact(ctx, job) + if err == nil || !strings.Contains(err.Error(), "digest mismatch") { + t.Fatalf("move error %v, want the agent's refusal", err) + } + if got, want := fleet.steps(), "check,pull,discard"; got != want { + t.Fatalf("steps %s, want %s", got, want) + } + waitUnpinned(t, ctx, client, job.PinIDs()) +} + +// TestMoveDiscardsAfterRetriesRunOut: a peer that never answers is retried +// under the policy, then the partial copy is discarded. +func TestMoveDiscardsAfterRetriesRunOut(t *testing.T) { + errs := make([]error, fastPolicy.MaxAttempts) + for i := range errs { + errs[i] = errors.New("peer unreachable") + } + fleet := &moveFleet{pullErrs: errs} + client := serveFleet(t, fleet, fastPolicy) + job := testJob("art-unreachable") + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + if _, err := client.MoveArtifact(ctx, job); err == nil { + t.Fatal("move succeeded against a peer that never answers") + } + want := "check," + strings.Repeat("pull,", fastPolicy.MaxAttempts) + "discard" + if got := fleet.steps(); got != want { + t.Fatalf("steps %s, want %s", got, want) + } + waitUnpinned(t, ctx, client, job.PinIDs()) +} + +// TestMoveFindsCommittedCopy: a copy already recorded on the destination is +// returned without a pull. +func TestMoveFindsCommittedCopy(t *testing.T) { + job := testJob("art-done") + fleet := &moveFleet{committed: job.LocalID} + client := serveFleet(t, fleet, fastPolicy) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + got, err := client.MoveArtifact(ctx, job) + if err != nil || got != job.LocalID { + t.Fatalf("move = %q, %v; want %q", got, err, job.LocalID) + } + if got := fleet.steps(); got != "check" { + t.Fatalf("steps %s, want check", got) + } +} + +// TestMoveSurvivesEndpointRestart stands in for kill -9 of the orchestrator +// mid-pull: the handler endpoint drops every connection while the pull runs, +// and a new endpoint on the same address takes the move over. the pins are +// restate state, so the restarted process still sees them held, and the move +// completes once. +func TestMoveSurvivesEndpointRestart(t *testing.T) { + first := &moveFleet{block: make(chan struct{}), pulling: make(chan struct{}, 1)} + defer close(first.block) + client, addr, stop := serveFleetAt(t, first, "") + job := testJob("art-restart") + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + done := make(chan error, 1) + var got string + go func() { + var err error + got, err = client.MoveArtifact(ctx, job) + done <- err + }() + + <-first.pulling + stop() + + second := &moveFleet{block: make(chan struct{}), pulling: make(chan struct{}, 1)} + _, _, _ = serveFleetAt(t, second, addr) + <-second.pulling + for i, p := range pinState(t, ctx, client, job.PinIDs()) { + if !p { + t.Fatalf("%s not pinned after the restart", job.PinIDs()[i]) + } + } + close(second.block) + + if err := <-done; err != nil { + t.Fatalf("move: %v", err) + } + if got != job.LocalID { + t.Fatalf("move returned %q, want %q", got, job.LocalID) + } + // the restarted endpoint replays check and the pins from the journal and + // runs only the unfinished pull, then the record. + if got, want := second.steps(), "pull,record"; got != want { + t.Fatalf("steps after the restart %s, want %s", got, want) + } + waitUnpinned(t, ctx, client, job.PinIDs()) +} + +// serveFleetAt serves fleet on addr ("" picks one) with an endpoint whose +// connections stop() drops at once, the way a killed process drops them. +func serveFleetAt(t *testing.T, fleet Fleet, addr string) (*Client, string, func()) { + t.Helper() + admin, ingressURL := restateTestURLs(t) + if addr == "" { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + addr = ln.Addr().String() + _ = ln.Close() + } + cfg := Config{Mode: ModeSelf, Listen: addr, IngressURL: ingressURL, AdminURL: admin, + Retry: map[string]RetryPolicy{ArtifactsObject: fastPolicy}} + h, err := Handler(cfg, fleet, testLogger) + if err != nil { + t.Fatal(err) + } + var protocols http.Protocols + protocols.SetUnencryptedHTTP2(true) + protocols.SetHTTP1(true) + srv := &http.Server{Handler: h, Protocols: &protocols} + ln, err := net.Listen("tcp", addr) + if err != nil { + t.Fatal(err) + } + go func() { _ = srv.Serve(ln) }() + t.Cleanup(func() { _ = srv.Close() }) + + client := NewClient(cfg) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if err := client.Register(ctx, "http://"+addr); err != nil { + t.Fatalf("register: %v", err) + } + return client, addr, func() { _ = srv.Close() } +} + +// TestPinLapsesThroughRestate: a pin nothing releases lapses on its timer, and +// a released one is gone at once while another holder's stays. +func TestPinLapsesThroughRestate(t *testing.T) { + client := serveFleet(t, fakeFleet{}, fastPolicy) + id := fmt.Sprintf("art-pin-%d", time.Now().UnixNano()) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + if err := client.PinArtifacts(ctx, "a", time.Hour, id); err != nil { + t.Fatal(err) + } + if err := client.PinArtifacts(ctx, "b", time.Second, id); err != nil { + t.Fatal(err) + } + if err := client.ReleaseArtifacts(ctx, "a", id); err != nil { + t.Fatal(err) + } + if p := pinState(t, ctx, client, []string{id}); !p[0] { + t.Fatal("released one holder and lost the other's pin") + } + waitUnpinned(t, ctx, client, []string{id}) +} diff --git a/internal/durable/client.go b/internal/durable/client.go index 7640afe7..bf9d1e23 100644 --- a/internal/durable/client.go +++ b/internal/durable/client.go @@ -22,6 +22,10 @@ type Client struct { apiKey string http *http.Client ingress *ingress.Client + // longIngress has no request timeout, for calls that wait on an + // invocation as long as its work takes (an artifact move). the caller's + // ctx bounds them. + longIngress *ingress.Client } // NewClient returns a client for cfg's ingress. a non-empty cfg.APIKey is sent @@ -30,17 +34,23 @@ func NewClient(cfg Config) *Client { // the transport carries the request id and trace context, so restate's // invocation joins the api request's trace and the handler sees the id. hc := &http.Client{Timeout: 30 * time.Second, Transport: &telemetry.Transport{Peer: "restate"}} + long := &http.Client{Transport: &telemetry.Transport{Peer: "restate"}} + return &Client{ + url: cfg.IngressURL, + adminURL: cfg.AdminURL, + apiKey: cfg.APIKey, + http: hc, + ingress: ingress.NewClient(cfg.IngressURL, ingressOptions(cfg, hc)...), + longIngress: ingress.NewClient(cfg.IngressURL, ingressOptions(cfg, long)...), + } +} + +func ingressOptions(cfg Config, hc *http.Client) []ingress.ClientOption { opts := []ingress.ClientOption{ingress.WithHttpClient(hc)} if cfg.APIKey != "" { opts = append(opts, ingress.WithAuthKey(cfg.APIKey)) } - return &Client{ - url: cfg.IngressURL, - adminURL: cfg.AdminURL, - apiKey: cfg.APIKey, - http: hc, - ingress: ingress.NewClient(cfg.IngressURL, opts...), - } + return opts } // authorize adds the api key, when there is one. diff --git a/internal/durable/destroy_test.go b/internal/durable/destroy_test.go index 343fce21..6574ce27 100644 --- a/internal/durable/destroy_test.go +++ b/internal/durable/destroy_test.go @@ -101,11 +101,7 @@ func (f *scriptedFleet) DestroyFailed(_ context.Context, vm, _, lastErr string) // registers it, and returns a client. skips without restate-server. func serveFleet(t *testing.T, fleet Fleet, policy RetryPolicy) *Client { t.Helper() - admin := os.Getenv("FUSE_RESTATE_TEST_ADMIN_URL") - ingressURL := os.Getenv("FUSE_RESTATE_TEST_INGRESS_URL") - if admin == "" || ingressURL == "" { - t.Skip("set FUSE_RESTATE_TEST_ADMIN_URL and FUSE_RESTATE_TEST_INGRESS_URL to run against restate-server") - } + admin, ingressURL := restateTestURLs(t) host := os.Getenv("FUSE_RESTATE_TEST_HANDLER_HOST") if host == "" { host = "127.0.0.1" @@ -118,7 +114,7 @@ func serveFleet(t *testing.T, fleet Fleet, policy RetryPolicy) *Client { _ = ln.Close() cfg := Config{Mode: ModeSelf, Listen: addr, IngressURL: ingressURL, AdminURL: admin, - Retry: map[string]RetryPolicy{EnvironmentObject: policy, HostObject: DefaultRetryPolicy}} + Retry: map[string]RetryPolicy{EnvironmentObject: policy, HostObject: DefaultRetryPolicy, ArtifactsObject: policy}} ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) go func() { _ = NewEndpoint(cfg, fleet, testLogger).Start(ctx, addr) }() @@ -131,6 +127,18 @@ func serveFleet(t *testing.T, fleet Fleet, policy RetryPolicy) *Client { return client } +// restateTestURLs returns the admin and ingress urls of the restate-server +// named by the test env, and skips without one. +func restateTestURLs(t *testing.T) (admin, ingressURL string) { + t.Helper() + admin = os.Getenv("FUSE_RESTATE_TEST_ADMIN_URL") + ingressURL = os.Getenv("FUSE_RESTATE_TEST_INGRESS_URL") + if admin == "" || ingressURL == "" { + t.Skip("set FUSE_RESTATE_TEST_ADMIN_URL and FUSE_RESTATE_TEST_INGRESS_URL to run against restate-server") + } + return admin, ingressURL +} + var fastPolicy = RetryPolicy{InitialInterval: 20 * time.Millisecond, MaxInterval: 50 * time.Millisecond, MaxAttempts: 4} // TestDestroyThroughRestate: a host that fails twice is retried inside the diff --git a/internal/durable/server.go b/internal/durable/server.go index ed727e88..23217961 100644 --- a/internal/durable/server.go +++ b/internal/durable/server.go @@ -22,10 +22,16 @@ const ( EnvironmentObject = "Environment" // HostObject is keyed by host id, for allocation and cordon. HostObject = "Host" + // ArtifactsObject is keyed by the snapshot id a move commits on its + // destination, so one move into a given copy runs at a time. + ArtifactsObject = "Artifacts" + // ArtifactPinObject is keyed by snapshot id and holds the pins that keep + // artifact gc off it. + ArtifactPinObject = "ArtifactPin" ) // services is every service this endpoint registers, in a fixed order. -var services = []string{EnvironmentObject, HostObject} +var services = []string{EnvironmentObject, HostObject, ArtifactsObject, ArtifactPinObject} // Fleet is all the handlers may see of the FleetManager. kept narrow on // purpose: handlers go through these calls, never into the manager's vm map, @@ -35,6 +41,7 @@ type Fleet interface { GetVM(vmID string) (orchestrator.VMInfo, bool) GetHost(hostID string) (orchestrator.Host, bool) Destroyer + Artifacts } // PingResult is what Ping returns. it never carries more than whether the @@ -70,10 +77,36 @@ func NewEndpoint(cfg Config, fleet Fleet, logger *slog.Logger) *server.Restate { }) })) + artifacts := restate.NewObject(ArtifactsObject, retryOption(cfg, ArtifactsObject)). + Handler("Move", restate.NewObjectHandler( + func(ctx restate.ObjectContext, job orchestrator.ArtifactMoveJob) (string, error) { + return move(ctx, fleet, servicePolicy(cfg, ArtifactsObject), job) + }, restate.WithAbortTimeout(moveAbortTimeout(fleet.ArtifactPullTimeout())))) + + pins := restate.NewObject(ArtifactPinObject, retryOption(cfg, ArtifactPinObject)). + Handler("Pin", restate.NewObjectHandler( + func(ctx restate.ObjectContext, req PinRequest) (restate.Void, error) { + return restate.Void{}, pin(ctx, req) + })). + Handler("Release", restate.NewObjectHandler( + func(ctx restate.ObjectContext, holder string) (restate.Void, error) { + return restate.Void{}, release(ctx, holder) + })). + Handler("Expire", restate.NewObjectHandler( + func(ctx restate.ObjectContext, e pinExpiry) (restate.Void, error) { + return restate.Void{}, expire(ctx, e) + })). + Handler("Pinned", restate.NewObjectSharedHandler( + func(ctx restate.ObjectSharedContext, _ restate.Void) (bool, error) { + return pinned(ctx) + })) + srv := server.NewRestate(). WithLogger(logger.Handler(), true). Bind(env). - Bind(host) + Bind(host). + Bind(artifacts). + Bind(pins) // a request without a valid signature from this key gets a 401 before // any handler runs. self mode leaves it unset: the endpoint is loopback. if cfg.IdentityKey != "" { diff --git a/internal/durable/server_test.go b/internal/durable/server_test.go index 23f913ed..1ea80a7d 100644 --- a/internal/durable/server_test.go +++ b/internal/durable/server_test.go @@ -43,6 +43,18 @@ func (fakeFleet) DestroyUnpublishIngress(context.Context, string) error { func (fakeFleet) DestroyFinish(context.Context, string, string) error { return nil } func (fakeFleet) DestroyFailed(context.Context, string, string, string) error { return nil } +func (fakeFleet) ArtifactMoveCommitted(context.Context, orchestrator.ArtifactMoveJob) (string, error) { + return "", nil +} +func (fakeFleet) ArtifactPull(context.Context, orchestrator.ArtifactMoveJob) (orchestrator.ArtifactMoved, error) { + return orchestrator.ArtifactMoved{}, nil +} +func (fakeFleet) ArtifactRecord(_ context.Context, job orchestrator.ArtifactMoveJob, _ orchestrator.ArtifactMoved) (string, error) { + return job.LocalID, nil +} +func (fakeFleet) ArtifactDiscard(context.Context, orchestrator.ArtifactMoveJob) error { return nil } +func (fakeFleet) ArtifactPullTimeout() time.Duration { return 0 } + var testLogger = slog.New(slog.NewTextHandler(io.Discard, nil)) // discovery is the manifest restate-server reads on registration. checking it @@ -99,8 +111,13 @@ func TestDiscoveryManifest(t *testing.T) { handlers[h.Name] = h.Ty } want := map[string]string{"Ping": "SHARED"} - if s.Name == EnvironmentObject { + switch s.Name { + case EnvironmentObject: want["Destroy"] = "EXCLUSIVE" + case ArtifactsObject: + want = map[string]string{"Move": "EXCLUSIVE"} + case ArtifactPinObject: + want = map[string]string{"Pin": "EXCLUSIVE", "Release": "EXCLUSIVE", "Expire": "EXCLUSIVE", "Pinned": "SHARED"} } if fmt.Sprint(handlers) != fmt.Sprint(want) { t.Errorf("%s: handlers %v, want %v", s.Name, handlers, want) diff --git a/orchestrator/main.go b/orchestrator/main.go index d794a1bc..3726e76b 100644 --- a/orchestrator/main.go +++ b/orchestrator/main.go @@ -254,7 +254,8 @@ func run() error { "largest startup script bound a request may ask for; must stay under -write-timeout") flag.DurationVar(&artifactPullTimeout, "artifact-pull-timeout", time.Duration(envInt("ORCH_ARTIFACT_PULL_TIMEOUT_SECONDS", 0))*time.Second, - "bound on one host-to-host artifact copy (0 = bound only by the request context)") + "bound on one attempt at a host-to-host artifact copy; with restate on it is the pull step's timeout, "+ + "not the request's (0 = bound only by the request context, or the agent client's timeout on restate)") flag.DurationVar(&checkpointInterval, "checkpoint-interval", time.Duration(envInt("ORCH_CHECKPOINT_INTERVAL_SECONDS", 0))*time.Second, "take a background diff checkpoint of every firecracker vm this often and keep it merged on "+ @@ -569,6 +570,7 @@ func run() error { if durableCfg.Mode != durable.ModeOff { restateClient = durable.NewClient(durableCfg) fm.SetDurableDestroyer(restateClient) + fm.SetDurableArtifacts(restateClient) } fm.Start(rootCtx) defer fm.Stop() From 752cb3e89c15235a9f6cf1344673cb0742019562 Mon Sep 17 00:00:00 2001 From: andrewn6 <59238070+andrewn6@users.noreply.github.com> Date: Sat, 14 Nov 2026 17:45:38 -0500 Subject: [PATCH 3/6] feat: meter host-to-host artifact copies by bytes, duration, hosts and failures --- internal/metrics/fleet_collector_test.go | 15 ++++++++ internal/metrics/metrics.go | 47 ++++++++++++++++++++++-- 2 files changed, 59 insertions(+), 3 deletions(-) diff --git a/internal/metrics/fleet_collector_test.go b/internal/metrics/fleet_collector_test.go index ad38ab82..e51e78de 100644 --- a/internal/metrics/fleet_collector_test.go +++ b/internal/metrics/fleet_collector_test.go @@ -105,3 +105,18 @@ func TestOperationCompletedCountsByResult(t *testing.T) { t.Fatalf("duration series = %d, want 1", n) } } + +func TestArtifactMoveCompletedCountsBytesAndFailures(t *testing.T) { + m := NewPrometheusMetrics(prometheus.NewRegistry()) + m.ArtifactMoveCompleted("h-a", "h-b", 4096, time.Second, nil) + m.ArtifactMoveCompleted("h-a", "h-b", 0, time.Second, errors.New("peer reset")) + if n := testutil.ToFloat64(m.artifactMoveBytes.WithLabelValues("h-a", "h-b")); n != 4096 { + t.Fatalf("bytes = %v, want 4096", n) + } + if n := testutil.ToFloat64(m.artifactMoveFailures.WithLabelValues("h-a", "h-b")); n != 1 { + t.Fatalf("failures = %v, want 1", n) + } + if n := testutil.CollectAndCount(m.artifactMoveDuration); n != 2 { + t.Fatalf("duration series = %d, want 2 (ok and error)", n) + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 7bfce7a7..67bdc06c 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -44,6 +44,11 @@ type PrometheusMetrics struct { // DestroyDeadLetterMetrics. destroyDeadLetters prometheus.Counter + // host-to-host artifact copies, through ArtifactMoveMetrics. + artifactMoveDuration *prometheus.HistogramVec + artifactMoveBytes *prometheus.CounterVec + artifactMoveFailures *prometheus.CounterVec + // lifecycle operations, through OperationMetrics. operations *prometheus.CounterVec operationDuration *prometheus.HistogramVec @@ -179,6 +184,26 @@ func NewPrometheusMetrics(reg prometheus.Registerer) *PrometheusMetrics { Help: "Total durable destroys that ran out of retries and left their environment in destroy_failed.", }), + artifactMoveDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: "orchestrator", + Subsystem: "artifact_move", + Name: "duration_seconds", + Help: "Host-to-host artifact copy attempts by source host, target host and result (ok, error).", + Buckets: []float64{.1, .5, 1, 5, 10, 30, 60, 120, 300, 600, 1800}, + }, []string{"source_host", "target_host", "result"}), + artifactMoveBytes: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "orchestrator", + Subsystem: "artifact_move", + Name: "bytes_total", + Help: "Bytes committed by successful host-to-host artifact copies, by source and target host.", + }, []string{"source_host", "target_host"}), + artifactMoveFailures: prometheus.NewCounterVec(prometheus.CounterOpts{ + Namespace: "orchestrator", + Subsystem: "artifact_move", + Name: "failures_total", + Help: "Failed host-to-host artifact copy attempts, by source and target host.", + }, []string{"source_host", "target_host"}), + operations: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: "orchestrator", Subsystem: "operations", @@ -250,6 +275,9 @@ func NewPrometheusMetrics(reg prometheus.Registerer) *PrometheusMetrics { m.egressProvisionFailed, m.egressTeardownFailed, m.destroyDeadLetters, + m.artifactMoveDuration, + m.artifactMoveBytes, + m.artifactMoveFailures, m.operations, m.operationDuration, m.stepDuration, @@ -299,6 +327,18 @@ func (m *PrometheusMetrics) DestroyDeadLettered() { m.destroyDeadLetters.Inc() } +// ArtifactMoveCompleted implements orchestrator.ArtifactMoveMetrics. +func (m *PrometheusMetrics) ArtifactMoveCompleted(sourceHostID, targetHostID string, bytes int64, d time.Duration, err error) { + result := "ok" + if err != nil { + result = "error" + m.artifactMoveFailures.WithLabelValues(sourceHostID, targetHostID).Inc() + } else if bytes > 0 { + m.artifactMoveBytes.WithLabelValues(sourceHostID, targetHostID).Add(float64(bytes)) + } + m.artifactMoveDuration.WithLabelValues(sourceHostID, targetHostID, result).Observe(d.Seconds()) +} + // EgressTeardownFailed implements orchestrator.EgressMetrics. func (m *PrometheusMetrics) EgressTeardownFailed(provider string) { m.egressTeardownFailed.WithLabelValues(provider).Inc() @@ -325,7 +365,8 @@ func (m *PrometheusMetrics) StepRetried(op, step string) { } var ( - _ orchestrator.EgressMetrics = (*PrometheusMetrics)(nil) - _ orchestrator.OperationMetrics = (*PrometheusMetrics)(nil) - _ durable.StepMetrics = (*PrometheusMetrics)(nil) + _ orchestrator.EgressMetrics = (*PrometheusMetrics)(nil) + _ orchestrator.OperationMetrics = (*PrometheusMetrics)(nil) + _ orchestrator.ArtifactMoveMetrics = (*PrometheusMetrics)(nil) + _ durable.StepMetrics = (*PrometheusMetrics)(nil) ) From 6c0ad7fbe761d3b4a09a1970b13ff2af9e2a459d Mon Sep 17 00:00:00 2001 From: andrewn6 <59238070+andrewn6@users.noreply.github.com> Date: Sat, 14 Nov 2026 17:45:38 -0500 Subject: [PATCH 4/6] docs: durable artifact moves, pull pins, move metrics and the pull timeout --- docs/content/docs/cloud/monitoring.mdx | 7 +++++++ .../docs/cloud/orchestrator-deploy.mdx | 3 ++- docs/runbooks/restate.mdx | 20 +++++++++++++++++++ 3 files changed, 29 insertions(+), 1 deletion(-) diff --git a/docs/content/docs/cloud/monitoring.mdx b/docs/content/docs/cloud/monitoring.mdx index 7e084c3d..e0b9682a 100644 --- a/docs/content/docs/cloud/monitoring.mdx +++ b/docs/content/docs/cloud/monitoring.mdx @@ -119,6 +119,13 @@ restate, one sample per executed step. A rising retry rate on one step is a host agent or dependency failing while restate keeps retrying it. See [Tracing](/docs/cloud/tracing) for following one request through its steps. +**Artifact moves**: `orchestrator_artifact_move_duration_seconds{source_host,target_host,result}`, +`orchestrator_artifact_move_bytes_total{source_host,target_host}` and +`orchestrator_artifact_move_failures_total{source_host,target_host}` record each +attempt at a host-to-host artifact copy (create seeding, off-host fork, +migrate, checkpoint deltas), with restate on or off. Failures concentrated on +one host pair point at the network between them or one agent. + **HTTP health**: the standard RED metrics apply directly. - `rate(orchestrator_http_requests_total{code=~"5.."}[5m])` for server error diff --git a/docs/content/docs/cloud/orchestrator-deploy.mdx b/docs/content/docs/cloud/orchestrator-deploy.mdx index 8479d046..05657a50 100644 --- a/docs/content/docs/cloud/orchestrator-deploy.mdx +++ b/docs/content/docs/cloud/orchestrator-deploy.mdx @@ -65,12 +65,13 @@ range covers the fleet. | `ORCH_ALLOWED_CIDRS` | `--allowed-cidrs` | empty → open | Comma-separated CIDR allowlist. | | `ORCH_SHUTDOWN_TIMEOUT_SECONDS` | `--shutdown-timeout` | `30` | Graceful shutdown ceiling. | | `ORCH_DRAINING_TIMEOUT_SECONDS` | `--draining-timeout` | `330` (5m30s) | Destroy an environment left `draining` this long with no `DELETE`. See [The reconcile loop](/docs/concepts/reconcile#stuck-draining). | +| `ORCH_ARTIFACT_PULL_TIMEOUT_SECONDS` | `--artifact-pull-timeout` | `0` | Bound on one attempt at a host-to-host artifact copy. With restate off it bounds the copy inside the request; with restate on it is the timeout of the `Artifacts` move's pull step, which no longer ends with the request. `0` leaves only the request (restate off) or the host agent client's own timeout (restate on). | | `ORCH_VM_PREFIX` | `--vm-prefix` | `fuse-` | Prefix applied to generated VM names. | | `FUSE_RESTATE_MODE` | | `off` | `off`, `self` or `cloud`. `self` serves restate handlers and checks restate in `/ready`; `cloud` uses a Restate Cloud environment, see [Restate Cloud](/docs/cloud/restate-cloud). | | `FUSE_RESTATE_INGRESS_URL` | | | Restate ingress URL, required unless the mode is `off`. `http://127.0.0.1:18080` with `ops/restate/compose.yaml`. | | `FUSE_RESTATE_API_KEY` / `FUSE_RESTATE_IDENTITY_KEY` | | | Mode `cloud` only: the Cloud API key and the `publickeyv1_...` signing key. Other cloud settings are on [Restate Cloud](/docs/cloud/restate-cloud#settings). | | `ORCH_RESTATE_LISTEN` | | `127.0.0.1:9080` | Restate handler endpoint (h2c). Only restate-server, or the TLS proxy in front of it in cloud public mode, should reach it. | -| `FUSE_RESTATE_RETRY__INITIAL_INTERVAL` / `_MAX_INTERVAL` / `_MAX_ATTEMPTS` | | `500ms` / `30s` / `10` | Invocation retry policy per restate service; `` is `ENVIRONMENT` or `HOST`. | +| `FUSE_RESTATE_RETRY__INITIAL_INTERVAL` / `_MAX_INTERVAL` / `_MAX_ATTEMPTS` | | `500ms` / `30s` / `10` | Invocation retry policy per restate service; `` is `ENVIRONMENT`, `HOST`, `ARTIFACTS` or `ARTIFACTPIN`. The `ARTIFACTS` policy also bounds the retries of an artifact move's pull. | With `FIRECRACKER_BASE_URL` unset, the orchestrator runs the in-memory stub provider (fine for local dev, not for production). With `DATABASE_URL` unset, state is diff --git a/docs/runbooks/restate.mdx b/docs/runbooks/restate.mdx index bdadccd8..bee11d5f 100644 --- a/docs/runbooks/restate.mdx +++ b/docs/runbooks/restate.mdx @@ -132,6 +132,26 @@ its last error in the ui; one that ran out of retries left the environment in (`orchestrator_destroy_dead_letters_total`). fix the host, then `DELETE` the environment again. +artifact moves between hosts (seeding a create, an off-host fork, a migrate, +a checkpoint delta) run as `Artifacts//Move`. the +pull is retried under the `ARTIFACTS` policy with the host-agent idempotency +key, so a restart mid-copy resumes it instead of starting a second one. a pull +that fails for good deletes the partial copy on the destination. while a move +runs, every snapshot it reads or writes is pinned in `ArtifactPin/` state, which artifact gc reads before it deletes anything; a pin nothing +releases (a killed invocation) lapses after 24h, a lazy migrate's after 35m. +gc skips an artifact whose pin it cannot read, so with restate down nothing +is collected. + +```sh +q "SELECT service_key, key, value_utf8 FROM state WHERE service_name = 'ArtifactPin'" +``` + +copies are metered by `orchestrator_artifact_move_duration_seconds`, +`orchestrator_artifact_move_bytes_total` and +`orchestrator_artifact_move_failures_total`, labeled `source_host` and +`target_host`. + ## restate-server is down - the orchestrator keeps serving its api; `/ready` reports restate degraded. From 6e7ab48d59e850c51a817cfe1b97e964903017a1 Mon Sep 17 00:00:00 2001 From: andrewn6 <59238070+andrewn6@users.noreply.github.com> Date: Sat, 14 Nov 2026 17:56:15 -0500 Subject: [PATCH 5/6] fix: keep partial copies and checkpoint files while a durable move may still be running --- internal/orchestrator/artifacts_durable.go | 41 ++++++++++++++++--- .../orchestrator/artifacts_durable_test.go | 41 +++++++++++++++++++ internal/orchestrator/checkpoints.go | 14 ++++++- internal/orchestrator/migrate_delta_test.go | 41 +++++++++++++++++++ internal/orchestrator/state_store.go | 12 +++++- internal/orchestrator/state_store_postgres.go | 2 +- 6 files changed, 142 insertions(+), 9 deletions(-) diff --git a/internal/orchestrator/artifacts_durable.go b/internal/orchestrator/artifacts_durable.go index 24e17770..6fdb6d54 100644 --- a/internal/orchestrator/artifacts_durable.go +++ b/internal/orchestrator/artifacts_durable.go @@ -59,12 +59,20 @@ func containsString(list []string, s string) bool { return false } +// ErrArtifactMoveUnsettled is a durable move the caller stopped waiting on +// without a result: the wait timed out or lost its connection to restate. the +// move may still be running, so nothing it reads or writes may be cleaned up +// on the strength of this error. +var ErrArtifactMoveUnsettled = errors.New("artifact move still running") + // DurableArtifacts runs artifact moves as the restate Artifacts/{local_id}.Move // handler and keeps pull pins in restate state, so neither is lost when the // orchestrator restarts. *durable.Client satisfies it. type DurableArtifacts interface { // MoveArtifact runs the move and waits for it, returning the snapshot id - // committed on the destination. the move keeps running if ctx ends first. + // committed on the destination. the move keeps running if ctx ends first; + // an error that is not the move's own failure wraps + // ErrArtifactMoveUnsettled. MoveArtifact(ctx context.Context, job ArtifactMoveJob) (string, error) // PinArtifacts holds ids against gc for holder until ReleaseArtifacts, or // until ttl passes. @@ -195,9 +203,9 @@ func (fm *FleetManager) ArtifactMoveCommitted(ctx context.Context, job ArtifactM // a lazy copy is never recorded. return "", nil } - record, err := fm.getSnapshotRecord(ctx, job.LocalID) - if err != nil { - return "", nil + record, found, err := fm.lookupSnapshotRecord(ctx, job.LocalID) + if err != nil || !found { + return "", err } if record.HostID == job.HostID && record.State == SnapshotStateReady { return record.SnapshotID, nil @@ -205,6 +213,22 @@ func (fm *FleetManager) ArtifactMoveCommitted(ctx context.Context, job ArtifactM return "", nil } +// lookupSnapshotRecord reads a snapshot record, telling a missing row +// (found false, nil error) apart from a failed read (an error). +func (fm *FleetManager) lookupSnapshotRecord(ctx context.Context, snapshotID string) (SnapshotRecord, bool, error) { + if fm.store == nil { + return SnapshotRecord{}, false, nil + } + record, err := fm.store.GetSnapshot(ctx, snapshotID) + if errors.Is(err, ErrSnapshotNotFound) { + return SnapshotRecord{}, false, nil + } + if err != nil { + return SnapshotRecord{}, false, fmt.Errorf("read snapshot %s: %w", snapshotID, err) + } + return record, true, nil +} + // errArtifactKindMismatch is a destination that committed something other // than what was sent: an agent too old for live snapshots or delta merges. // retrying cannot change it. @@ -405,9 +429,14 @@ func (fm *FleetManager) recordMergedDelta(ctx context.Context, job ArtifactMoveJ // ArtifactDiscard deletes what a failed move may have left on the // destination. a copy that did get recorded is kept: the record step can -// commit on an attempt whose answer was lost. +// commit on an attempt whose answer was lost. a record that cannot be read +// fails the step, so the bytes stay until the row is known to be absent. func (fm *FleetManager) ArtifactDiscard(ctx context.Context, job ArtifactMoveJob) error { - if record, err := fm.getSnapshotRecord(ctx, job.LocalID); err == nil && record.HostID == job.HostID { + record, found, err := fm.lookupSnapshotRecord(ctx, job.LocalID) + if err != nil { + return err + } + if found && record.HostID == job.HostID { return nil } if err := fm.deleteFreeArtifact(ctx, job.HostID, job.LocalID); err != nil && !IsNotFound(err) { diff --git a/internal/orchestrator/artifacts_durable_test.go b/internal/orchestrator/artifacts_durable_test.go index 65f84d62..a3ca1d41 100644 --- a/internal/orchestrator/artifacts_durable_test.go +++ b/internal/orchestrator/artifacts_durable_test.go @@ -301,3 +301,44 @@ func TestArtifactPull_metrics(t *testing.T) { t.Fatalf("metrics %v", m.got) } } + +// flakySnapshotStore fails every snapshot read. +type flakySnapshotStore struct{ *MemoryStateStore } + +func (flakySnapshotStore) GetSnapshot(context.Context, string) (SnapshotRecord, error) { + return SnapshotRecord{}, errors.New("connection refused") +} + +// TestArtifactDiscard_unreadableRecordKeepsBytes: a record that cannot be +// read may exist, so the copy is not deleted and the step fails to retry. +func TestArtifactDiscard_unreadableRecordKeepsBytes(t *testing.T) { + ctx := context.Background() + stub := &deletingProvider{stubLoadProvider: newStubProvider()} + fm := NewFleetManager(FleetConfig{Provider: stub, Prefix: "fuse-", StateStore: flakySnapshotStore{NewMemoryStateStore()}}) + if err := fm.RegisterHost(ctx, artifactHost("h-b", 8), stub); err != nil { + t.Fatal(err) + } + if err := fm.ArtifactDiscard(ctx, ArtifactMoveJob{HostID: "h-b", LocalID: "art-x"}); err == nil { + t.Fatal("discard succeeded with the record unreadable") + } + if len(stub.deleted) != 0 { + t.Fatalf("deleted %v with the record unreadable", stub.deleted) + } + if _, err := fm.ArtifactMoveCommitted(ctx, ArtifactMoveJob{HostID: "h-b", LocalID: "art-x"}); err == nil { + t.Fatal("commit check hid a failed read") + } +} + +// TestEnsureArtifactOnHost_unsettledIsDistinct: a durable move the caller +// stopped waiting on stays recognisable, so cleanup can leave its files. +func TestEnsureArtifactOnHost_unsettledIsDistinct(t *testing.T) { + ctx := context.Background() + fm, _ := twoHostFleet(t, &recordingMover{}) + d := newFakeDurableArtifacts() + d.moveErr = fmt.Errorf("%w: context deadline exceeded", ErrArtifactMoveUnsettled) + fm.SetDurableArtifacts(d) + _, err := fm.ensureArtifactOnHost(ctx, seedArtifactRecord(t, fm, "art-1", "tenant-1", "h-a", "d1"), "h-b", false) + if !errors.Is(err, ErrArtifactMoveUnsettled) || !errors.Is(err, ErrArtifactTransferFailed) { + t.Fatalf("err %v, want it to stay unsettled", err) + } +} diff --git a/internal/orchestrator/checkpoints.go b/internal/orchestrator/checkpoints.go index c56877dc..433b1231 100644 --- a/internal/orchestrator/checkpoints.go +++ b/internal/orchestrator/checkpoints.go @@ -2,6 +2,7 @@ package orchestrator import ( "context" + "errors" "sort" "sync" "time" @@ -362,6 +363,13 @@ func (fm *FleetManager) advanceChain(ctx context.Context, vmID string) { if err != nil { fm.logger.Warn("move checkpoint delta failed; chain reset", "vm", vmID, "host", c.standby, "err", err) fm.appendEventBackground("vm", vmID, "vm.checkpoint_chain_reset", map[string]any{"error": err.Error()}) + if errors.Is(err, ErrArtifactMoveUnsettled) { + // the merge may still be reading the child and writing onto + // the base, so neither is deleted: the chain is only + // forgotten. retention gc takes the child later. + fm.resetChainLocked(c) + return + } // the base may or may not have been merged and dropped; either way // the chain no longer has an agreed state. fm.deleteCheckpoint(ctx, child) @@ -397,7 +405,11 @@ func (fm *FleetManager) advanceChain(ctx context.Context, vmID string) { base, err := fm.ensureArtifactOnHost(ctx, full, standby, false) if err != nil { fm.logger.Warn("copy full checkpoint to standby failed", "vm", vmID, "host", standby, "err", err) - fm.deleteCheckpoint(ctx, full) + if !errors.Is(err, ErrArtifactMoveUnsettled) { + // an unsettled copy may still be reading it; retention gc takes + // it later instead. + fm.deleteCheckpoint(ctx, full) + } return } fm.beginArtifactPull(base) diff --git a/internal/orchestrator/migrate_delta_test.go b/internal/orchestrator/migrate_delta_test.go index aff5457e..fa1f27b5 100644 --- a/internal/orchestrator/migrate_delta_test.go +++ b/internal/orchestrator/migrate_delta_test.go @@ -353,3 +353,44 @@ func TestCheckpointChain_aRefusedDiffStartsOver(t *testing.T) { t.Errorf("moves = %+v, want the restart to copy a fresh full checkpoint", moves) } } + +// TestCheckpointChain_anUnsettledCopyKeepsItsFiles: a durable copy the loop +// stopped waiting on may still be running, so the checkpoint it reads stays +// recorded and the base it writes is not deleted. +func TestCheckpointChain_anUnsettledCopyKeepsItsFiles(t *testing.T) { + provider := newDeltaMigrateProvider() + fm := newDeltaFleet(t, provider, &deltaMover{}) + d := newFakeDurableArtifacts() + d.moveErr = fmt.Errorf("%w: context deadline exceeded", ErrArtifactMoveUnsettled) + fm.SetDurableArtifacts(d) + ctx := context.Background() + + fm.advanceChain(ctx, "fuse-task-src") + if len(d.jobs) != 1 { + t.Fatalf("%d durable moves, want the full copy", len(d.jobs)) + } + if _, err := fm.GetSnapshotByID(ctx, d.jobs[0].SnapshotID); err != nil { + t.Fatalf("full checkpoint %s deleted while its copy may still run: %v", d.jobs[0].SnapshotID, err) + } + + // a settled copy, then an unsettled delta: the chain is forgotten, its + // child and base left alone. + d.moveErr = nil + fm.advanceChain(ctx, "fuse-task-src") + c := fm.chainFor("fuse-task-src") + head := c.head.SnapshotID + d.moveErr = fmt.Errorf("%w: connection reset", ErrArtifactMoveUnsettled) + fm.advanceChain(ctx, "fuse-task-src") + child := d.jobs[len(d.jobs)-1].SnapshotID + if c.base != "" || c.head.SnapshotID != "" { + t.Errorf("chain = head %s base %s, want it forgotten", c.head.SnapshotID, c.base) + } + if _, err := fm.GetSnapshotByID(ctx, child); err != nil { + t.Errorf("delta child %s deleted while its merge may still run: %v", child, err) + } + // retiring the chain would have deleted the head the delta is taken + // against, and the base on the standby with it. + if _, err := fm.GetSnapshotByID(ctx, head); err != nil { + t.Errorf("chain head %s deleted, so the chain was retired rather than forgotten: %v", head, err) + } +} diff --git a/internal/orchestrator/state_store.go b/internal/orchestrator/state_store.go index dc15f836..a2ed497c 100644 --- a/internal/orchestrator/state_store.go +++ b/internal/orchestrator/state_store.go @@ -556,13 +556,23 @@ func (s *MemoryStateStore) UpsertSnapshot(_ context.Context, snapshot SnapshotRe return nil } +// snapshotMissingError is a store's answer for a snapshot id it has no row +// for. it matches ErrSnapshotNotFound, so a caller can tell a missing row +// apart from a failed read. +type snapshotMissingError struct{ id string } + +func (e snapshotMissingError) Error() string { return "snapshot " + e.id + " not found" } +func (e snapshotMissingError) Is(target error) bool { return target == ErrSnapshotNotFound } + +func snapshotMissing(id string) error { return snapshotMissingError{id: id} } + func (s *MemoryStateStore) GetSnapshot(_ context.Context, snapshotID string) (SnapshotRecord, error) { s.mu.RLock() defer s.mu.RUnlock() snapshot, ok := s.snapshots[snapshotID] if !ok { - return SnapshotRecord{}, fmt.Errorf("snapshot %s not found", snapshotID) + return SnapshotRecord{}, snapshotMissing(snapshotID) } return snapshot, nil } diff --git a/internal/orchestrator/state_store_postgres.go b/internal/orchestrator/state_store_postgres.go index 9057ab86..11bd3a54 100644 --- a/internal/orchestrator/state_store_postgres.go +++ b/internal/orchestrator/state_store_postgres.go @@ -438,7 +438,7 @@ func (s *PostgresStateStore) GetSnapshot(ctx context.Context, snapshotID string) record, err := scanSnapshotRow(row.Scan) if err != nil { if err == sql.ErrNoRows { - return SnapshotRecord{}, fmt.Errorf("snapshot %s not found", snapshotID) + return SnapshotRecord{}, snapshotMissing(snapshotID) } return SnapshotRecord{}, fmt.Errorf("get snapshot %s: %w", snapshotID, err) } From 26e448d4d23726f9bb60133b49b7d554229329a0 Mon Sep 17 00:00:00 2001 From: andrewn6 <59238070+andrewn6@users.noreply.github.com> Date: Sat, 14 Nov 2026 17:56:15 -0500 Subject: [PATCH 6/6] fix: tell a failed move from an abandoned wait and hold move pins until the move releases them --- docs/runbooks/restate.mdx | 17 ++++++-- internal/durable/artifacts.go | 67 +++++++++++++++++++++++------- internal/durable/artifacts_test.go | 10 +++++ 3 files changed, 75 insertions(+), 19 deletions(-) diff --git a/docs/runbooks/restate.mdx b/docs/runbooks/restate.mdx index bee11d5f..f4e0ef37 100644 --- a/docs/runbooks/restate.mdx +++ b/docs/runbooks/restate.mdx @@ -138,10 +138,21 @@ pull is retried under the `ARTIFACTS` policy with the host-agent idempotency key, so a restart mid-copy resumes it instead of starting a second one. a pull that fails for good deletes the partial copy on the destination. while a move runs, every snapshot it reads or writes is pinned in `ArtifactPin/` state, which artifact gc reads before it deletes anything; a pin nothing -releases (a killed invocation) lapses after 24h, a lazy migrate's after 35m. +id>` state, which artifact gc reads before it deletes anything. a move's pins +have no timer: every way the move ends (done, failed, cancelled) releases +them, so only a killed invocation leaves them behind. release those by hand +with the holder the state query below shows: + +```sh +curl -s localhost:18080/ArtifactPin//Release \ + -H 'content-type: application/json' -d '"move:"' +``` + +a lazy migrate's seed pin lapses after 35m if its watcher never releases it. gc skips an artifact whose pin it cannot read, so with restate down nothing -is collected. +is collected. a caller that stops waiting on a move (a restart, or an hour +with no answer) leaves the files that move uses alone; the checkpoint loop +forgets its chain instead of deleting it. ```sh q "SELECT service_key, key, value_utf8 FROM state WHERE service_name = 'ArtifactPin'" diff --git a/internal/durable/artifacts.go b/internal/durable/artifacts.go index 3cb07dd2..119a1166 100644 --- a/internal/durable/artifacts.go +++ b/internal/durable/artifacts.go @@ -2,6 +2,8 @@ package durable import ( "context" + "errors" + "fmt" "time" restate "github.com/restatedev/sdk-go" @@ -13,12 +15,6 @@ import ( // opArtifactMove is the op label on a move's step spans and metrics. const opArtifactMove = "artifact_move" -// movePinTTL is how long a move's pins last if the move never releases them, -// as when its invocation is killed. a move that finishes, fails or is -// cancelled releases them itself; this only stops a dead one holding an -// artifact forever. -const movePinTTL = 24 * time.Hour - // defaultPullAttemptBound is how long one pull attempt can take with no // ORCH_ARTIFACT_PULL_TIMEOUT_SECONDS: the agent client's 10 minute request // timeout, three times over for its keyed retries, plus their backoff. @@ -51,7 +47,27 @@ func moveAbortTimeout(pullTimeout time.Duration) time.Duration { // state for the whole move, so gc leaves them alone across a restart. a pull // or record that fails for good deletes what reached the destination, so no // copy is left there that nothing records. +// +// the pins have no lapse timer: a move can outlast any fixed one while its +// pull retries, and every way it ends (done, failed, cancelled) releases them. +// only a killed invocation leaves them, for an operator to release. func move(ctx restate.ObjectContext, fleet Artifacts, policy RetryPolicy, job orchestrator.ArtifactMoveJob) (string, error) { + id, err := moveSteps(ctx, fleet, policy, job) + if err != nil && restate.IsTerminalError(err) { + // the code is what tells the waiting caller that the move itself + // ended, whichever restate-server version answers. + return "", restate.ToTerminalError(err, restate.WithErrorCode(moveFailedCode)) + } + return id, err +} + +// moveFailedCode is the error code a move that ended in failure answers its +// caller with. a caller that gets any other error stopped waiting without +// a result. not every restate-server marks an invocation's own failure with +// the x-restate-error-source header, so the code carries it instead. +const moveFailedCode restate.Code = 422 + +func moveSteps(ctx restate.ObjectContext, fleet Artifacts, policy RetryPolicy, job orchestrator.ArtifactMoveJob) (string, error) { if key := restate.Key(ctx); job.LocalID != key { return "", restate.TerminalErrorf("move into %s sent to Artifacts/%s", job.LocalID, key) } @@ -67,7 +83,7 @@ func move(ctx restate.ObjectContext, fleet Artifacts, policy RetryPolicy, job or ids := job.PinIDs() for _, id := range ids { if _, err := restate.Object[restate.Void](ctx, ArtifactPinObject, id, "Pin"). - Request(PinRequest{Holder: holder, TTL: movePinTTL}); err != nil { + Request(PinRequest{Holder: holder}); err != nil { return "", err } } @@ -114,6 +130,7 @@ func releasePins(ctx restate.Context, holder string, ids []string) { } // PinRequest holds an ArtifactPin for Holder, until released or TTL passes. +// a zero TTL never lapses. type PinRequest struct { Holder string `json:"holder"` TTL time.Duration `json:"ttl"` @@ -133,8 +150,8 @@ const pinHoldersKey = "holders" // than a sleeping invocation, so a pin held for hours pins nothing to this // deployment. func pin(ctx restate.ObjectContext, req PinRequest) error { - if req.Holder == "" || req.TTL <= 0 { - return restate.TerminalErrorf("pin needs a holder and a positive ttl") + if req.Holder == "" || req.TTL < 0 { + return restate.TerminalErrorf("pin needs a holder and a ttl that is not negative") } now, err := restate.Run(ctx, func(restate.RunContext) (int64, error) { return time.Now().UnixMilli(), nil @@ -149,6 +166,12 @@ func pin(ctx restate.ObjectContext, req PinRequest) error { if holders == nil { holders = map[string]int64{} } + if req.TTL == 0 { + // held until released. 0 is never a deadline, so no lapse matches it. + holders[req.Holder] = 0 + restate.Set(ctx, pinHoldersKey, holders) + return nil + } deadline := now + req.TTL.Milliseconds() holders[req.Holder] = deadline restate.Set(ctx, pinHoldersKey, holders) @@ -159,23 +182,23 @@ func pin(ctx restate.ObjectContext, req PinRequest) error { // release is ArtifactPin/{snapshot_id}.Release. func release(ctx restate.ObjectContext, holder string) error { - return dropHolder(ctx, holder, 0) + return dropHolder(ctx, holder, 0, false) } // expire is ArtifactPin/{snapshot_id}.Expire, the pin's timer. func expire(ctx restate.ObjectContext, e pinExpiry) error { - return dropHolder(ctx, e.Holder, e.Deadline) + return dropHolder(ctx, e.Holder, e.Deadline, true) } -// dropHolder removes holder, or with a deadline only the pin that deadline -// belongs to. the last holder out clears the object. -func dropHolder(ctx restate.ObjectContext, holder string, deadline int64) error { +// dropHolder removes holder (lapse false), or on a lapse only the pin that +// deadline belongs to. the last holder out clears the object. +func dropHolder(ctx restate.ObjectContext, holder string, deadline int64, lapse bool) error { holders, err := restate.Get[map[string]int64](ctx, pinHoldersKey) if err != nil { return err } got, ok := holders[holder] - if !ok || (deadline != 0 && got != deadline) { + if !ok || (lapse && (got == 0 || got != deadline)) { return nil } delete(holders, holder) @@ -201,8 +224,20 @@ func pinned(ctx restate.ObjectSharedContext) (bool, error) { // request timeout of its own; ctx bounds the wait, and the move keeps running // if ctx ends first. func (c *Client) MoveArtifact(ctx context.Context, job orchestrator.ArtifactMoveJob) (string, error) { - return ingress.Object[orchestrator.ArtifactMoveJob, string](c.longIngress, ArtifactsObject, job.LocalID, "Move"). + id, err := ingress.Object[orchestrator.ArtifactMoveJob, string](c.longIngress, ArtifactsObject, job.LocalID, "Move"). Request(ctx, job) + if err != nil && !moveEnded(err) { + // no answer from the move itself: it may still be running. + return "", fmt.Errorf("%w: %w", orchestrator.ErrArtifactMoveUnsettled, err) + } + return id, err +} + +// moveEnded reports whether err is the move's own failure, answered with +// moveFailedCode. +func moveEnded(err error) bool { + var ierr ingress.Error + return errors.As(err, &ierr) && ierr.Code() == moveFailedCode } // PinArtifacts pins each id for holder, for at most ttl. diff --git a/internal/durable/artifacts_test.go b/internal/durable/artifacts_test.go index 47e87681..6ac17d55 100644 --- a/internal/durable/artifacts_test.go +++ b/internal/durable/artifacts_test.go @@ -148,6 +148,13 @@ func TestMoveThroughRestate(t *testing.T) { t.Fatalf("%s not pinned during the pull", job.PinIDs()[i]) } } + // a caller that stops waiting gets an unsettled error, not a failure. + short, stop := context.WithTimeout(ctx, 200*time.Millisecond) + _, err := client.MoveArtifact(short, job) + stop() + if !errors.Is(err, orchestrator.ErrArtifactMoveUnsettled) { + t.Fatalf("abandoned wait: %v, want ErrArtifactMoveUnsettled", err) + } close(fleet.block) if err := <-done; err != nil { @@ -175,6 +182,9 @@ func TestMoveDiscardsAfterPermanentFailure(t *testing.T) { if err == nil || !strings.Contains(err.Error(), "digest mismatch") { t.Fatalf("move error %v, want the agent's refusal", err) } + if errors.Is(err, orchestrator.ErrArtifactMoveUnsettled) { + t.Fatalf("the move's own failure %v reads as unsettled", err) + } if got, want := fleet.steps(), "check,pull,discard"; got != want { t.Fatalf("steps %s, want %s", got, want) }