feat: run artifact moves as a durable restate service with pull pins in durable state

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

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_<SERVICE>_INITIAL_INTERVAL` / `_MAX_INTERVAL` / `_MAX_ATTEMPTS` | | `500ms` / `30s` / `10` | Invocation retry policy per restate service; `<SERVICE>` is `ENVIRONMENT` or `HOST`. |
+| `FUSE_RESTATE_RETRY_<SERVICE>_INITIAL_INTERVAL` / `_MAX_INTERVAL` / `_MAX_ATTEMPTS` | | `500ms` / `30s` / `10` | Invocation retry policy per restate service; `<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

docs/runbooks/restate.mdx

@@ -132,6 +132,37 @@ 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/<destination snapshot id>/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/<snapshot
+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/<snapshot id>/Release \
+  -H 'content-type: application/json' -d '"move:<invocation id>"'
+```
+
+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. 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'"
+```
+
+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.

internal/durable/artifacts.go

@@ -0,0 +1,271 @@
+package durable
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"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"
+
+// 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.
+//
+// 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)
+	}
+
+	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}); 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.
+// a zero TTL never lapses.
+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 ttl that is not negative")
+	}
+	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{}
+	}
+	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)
+	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, 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, true)
+}
+
+// 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 || (lapse && (got == 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) {
+	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.
+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)