feat: run token rotation as one restate journal with only ciphertext journaled

docs/content/docs/cloud/monitoring.mdx

@@ -103,7 +103,7 @@ reported rather than freezing at its last value:
 
 **Lifecycle operations**: `orchestrator_operations_total{op,result}` and
 `orchestrator_operations_duration_seconds{op}` cover `create`, `destroy`,
-`fork`, `migrate`, `migrate_live`, `snapshot` and `restore`. `destroy` counts
+`fork`, `migrate`, `migrate_live`, `snapshot`, `restore` and `rotate_token`. `destroy` counts
 explicit destroys, from the API or a migrate removing its source. Teardowns the
 reconcile loop does on its own (stuck, idle, orphaned) and the teardown after a
 task completes aren't included; the reconcile counters above cover the first

docs/content/docs/cloud/orchestrator-deploy.mdx

@@ -71,15 +71,15 @@ range covers the fleet.
 | `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`, `HOST`, `ARTIFACTS` or `ARTIFACTPIN`. The `ARTIFACTS` policy also bounds the retries of an artifact move's pull; the `ENVIRONMENT` policy bounds the host steps of a destroy, snapshot or restore, and the retries of a snapshot gc deletion. |
+| `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; the `ENVIRONMENT` policy bounds the host steps of a destroy, snapshot, restore or token rotation upload (a rotation's state store write retries without bound), and the retries of a snapshot gc deletion. |
 
 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
 in-memory and does not survive a restart.
 
 ## Restate
 
-Restate is optional and off by default. With it on, destroys, artifact moves, snapshots, restores, the checkpoint chain and snapshot gc run on it. To run it next to the orchestrator:
+Restate is optional and off by default. With it on, destroys, artifact moves, snapshots, restores, the checkpoint chain, snapshot gc and token rotation run on it. To run it next to the orchestrator:
 
 ```bash
 docker compose -f ops/restate/compose.yaml up -d

docs/runbooks/restate.mdx

@@ -186,6 +186,22 @@ retention gc marks an expired snapshot `deleting` and sends
 (`snapshot.gc_failed`); the next sweep marks it again. gc skips a snapshot any
 pin holds, in process or in `ArtifactPin`.
 
+token rotation runs as `Environment/<vm_id>/RotateToken`: generate, upload,
+persist. generate is journaled, so a restart between the upload and the
+persist writes the token already in the guest to the state store instead of
+making a second one. the upload retries under the `ENVIRONMENT` policy; the
+persist has no attempt bound, so a state store that stays down holds the
+rotation (and the vm's other environment ops) until it is back. the journaled value is ciphertext under
+`TOKEN_ENCRYPTION_KEY` (the token, cert and key), and the steps decrypt it
+themselves; no plaintext token reaches the journal. a rotation left unfinished
+across a `TOKEN_ENCRYPTION_KEY` change fails, since its ciphertext no longer
+decrypts; run it again. a rotation's journal can be read with the query below. rotations are counted as
+`orchestrator_operations_total{op="rotate_token"}`.
+
+```sh
+q "SELECT j.index, j.name, j.entry_json FROM sys_journal j JOIN sys_invocation i ON j.id = i.id WHERE i.target_handler_name = 'RotateToken'"
+```
+
 copies are metered by `orchestrator_artifact_move_duration_seconds`,
 `orchestrator_artifact_move_bytes_total` and
 `orchestrator_artifact_move_failures_total`, labeled `source_host` and

internal/durable/server.go

@@ -44,6 +44,7 @@ type Fleet interface {
 	Artifacts
 	Snapshots
 	Checkpoints
+	Tokens
 }
 
 // snapshotAbortTimeout is how long restate lets a snapshot, restore or
@@ -105,6 +106,10 @@ func NewEndpoint(cfg Config, fleet Fleet, logger *slog.Logger) *server.Restate {
 		Handler("RetireChain", restate.NewObjectHandler(
 			func(ctx restate.ObjectContext, _ restate.Void) (restate.Void, error) {
 				return restate.Void{}, dropChain(ctx, fleet)
+			})).
+		Handler("RotateToken", restate.NewObjectHandler(
+			func(ctx restate.ObjectContext, _ restate.Void) (restate.Void, error) {
+				return restate.Void{}, rotateToken(ctx, fleet, servicePolicy(cfg, EnvironmentObject))
 			}))
 
 	host := restate.NewObject(HostObject, retryOption(cfg, HostObject)).

internal/durable/server_test.go

@@ -99,6 +99,16 @@ func (fakeFleet) ChainAdvanced(context.Context, string, orchestrator.ChainState,
 }
 func (fakeFleet) MirrorChain(string, orchestrator.ChainState, bool) {}
 
+func (fakeFleet) RotateTokenGenerate(context.Context, string) (orchestrator.RotatedCredentials, error) {
+	return orchestrator.RotatedCredentials{}, nil
+}
+func (fakeFleet) RotateTokenUpload(context.Context, string, orchestrator.RotatedCredentials) error {
+	return nil
+}
+func (fakeFleet) RotateTokenPersist(context.Context, string, orchestrator.RotatedCredentials) error {
+	return nil
+}
+
 var testLogger = slog.New(slog.NewTextHandler(io.Discard, nil))
 
 // discovery is the manifest restate-server reads on registration. checking it
@@ -157,7 +167,7 @@ func TestDiscoveryManifest(t *testing.T) {
 		want := map[string]string{"Ping": "SHARED"}
 		switch s.Name {
 		case EnvironmentObject:
-			for _, h := range []string{"Destroy", "Snapshot", "Restore", "CollectSnapshot", "Checkpoint", "ChainMoved", "HoldChain", "ReleaseChain", "RetireChain"} {
+			for _, h := range []string{"Destroy", "Snapshot", "Restore", "CollectSnapshot", "Checkpoint", "ChainMoved", "HoldChain", "ReleaseChain", "RetireChain", "RotateToken"} {
 				want[h] = "EXCLUSIVE"
 			}
 		case ArtifactsObject:

internal/durable/token.go

@@ -0,0 +1,83 @@
+package durable
+
+import (
+	"context"
+	"errors"
+
+	restate "github.com/restatedev/sdk-go"
+	"github.com/restatedev/sdk-go/ingress"
+
+	"github.com/folsomintel/fuse/internal/orchestrator"
+)
+
+// opRotateToken is the op label on token rotation's step spans and metrics.
+const opRotateToken = "rotate_token"
+
+// rotateRefusedCode is the error code a token rotation's own refusal (vm not
+// running, no encryption key) crosses restate as, so the client hands the api
+// the fleet's message unchanged. 480-486 are the snapshot sentinels.
+const rotateRefusedCode restate.Code = 487
+
+// Tokens is the part of the fleet the token rotation handler drives, one
+// method per step. *FleetManager satisfies it.
+type Tokens interface {
+	RotateTokenGenerate(ctx context.Context, vmID string) (orchestrator.RotatedCredentials, error)
+	RotateTokenUpload(ctx context.Context, vmID string, rotated orchestrator.RotatedCredentials) error
+	RotateTokenPersist(ctx context.Context, vmID string, rotated orchestrator.RotatedCredentials) error
+}
+
+// rotatePermanent marks err terminal when a retry cannot change it: the
+// rotation's own refusals under rotateRefusedCode, the rest the way snapshot
+// steps mark theirs (vm not found keeps its sentinel, an agent 4xx is final).
+func rotatePermanent(err error) error {
+	if orchestrator.TokenRotationRefused(err) {
+		return restate.ToTerminalError(err, restate.WithErrorCode(rotateRefusedCode))
+	}
+	return permanent(err)
+}
+
+// rotateToken is Environment/{vm_id}.RotateToken: generate, upload, persist.
+// the upload retries under policy; the persist retries until it succeeds.
+// generate is journaled, so a replay after a crash uploads and persists the
+// token it already made instead of a second one, and the stored token ends up
+// the one in the guest. its journaled value is ciphertext only: the plaintext
+// token is decrypted inside the upload and persist steps and never returned.
+func rotateToken(ctx restate.ObjectContext, fleet Tokens, policy RetryPolicy) error {
+	vmID := restate.Key(ctx)
+	retry := retryUnder(policy)
+	rotated, err := Run(ctx, opRotateToken, "generate", func(c context.Context) (orchestrator.RotatedCredentials, error) {
+		rotated, err := fleet.RotateTokenGenerate(c, vmID)
+		return rotated, rotatePermanent(err)
+	})
+	if err != nil {
+		return err
+	}
+	if err := runStep(ctx, opRotateToken, "upload", func(c context.Context) error {
+		return rotatePermanent(fleet.RotateTokenUpload(c, vmID, rotated))
+	}, retry...); err != nil {
+		return err
+	}
+	// the guest holds the new token now, so the persist has no attempt
+	// bound: a state store that stays down holds the rotation here until it
+	// is back, instead of ending it with the old token stored.
+	return runStep(ctx, opRotateToken, "persist", func(c context.Context) error {
+		return rotatePermanent(fleet.RotateTokenPersist(c, vmID, rotated))
+	}, restate.WithInitialRetryInterval(policy.InitialInterval),
+		restate.WithMaxRetryInterval(policy.MaxInterval))
+}
+
+// RotateToken runs Environment/{vmID}.RotateToken and waits for it. the
+// rotation may queue behind a long snapshot on the same vm, so it goes through
+// the client with no request timeout of its own; ctx bounds the wait, and the
+// rotation keeps running if ctx ends first.
+func (c *Client) RotateToken(ctx context.Context, vmID string) error {
+	_, err := ingress.Object[restate.Void, restate.Void](c.longIngress, EnvironmentObject, vmID, "RotateToken").
+		Request(ctx, restate.Void{})
+	var ierr ingress.Error
+	if errors.As(err, &ierr) && ierr.Code() == rotateRefusedCode {
+		return errors.New(ierr.Message())
+	}
+	return fleetError(err)
+}
+
+var _ orchestrator.DurableTokenRotator = (*Client)(nil)