amadan
Merge pull request #353 from carlosframework/instance-moves
Instance moves: two-phase, data-following residence moves (#331)
24 files changed,
+2107
−83
docs/superpowers/plans/2026-08-28-instance-moves.md+109 −0docs/superpowers/specs/2026-08-28-instance-move-design.md+61 −35infra/modules/region-host/main.tf+19 −2infra/modules/region-host/variables.tf+6 −0infra/modules/region-remote-host/main.tf+18 −2infra/modules/region-remote-host/variables.tf+6 −0internal/activator/activator.go+12 −0internal/activator/backend_exec.go+104 −4internal/activator/backend_exec_test.go+142 −0internal/console/apiinstancemove.go+156 −0internal/console/apiinstancemove_test.go+184 −0internal/console/authcli.go+8 −0internal/console/instancesplacement.go+8 −6internal/console/svcroutes_test.go+13 −2internal/host/agent.go+14 −0internal/host/edgesfile.go+17 −6internal/host/edgesfile_test.go+65 −26internal/host/movespass.go+300 −0internal/host/movespass_test.go+294 −0internal/lease/mirror.go+165 −0internal/lease/mirror_test.go+112 −0internal/provision/provision.go+165 −0internal/provision/provision_test.go+108 −0internal/registry/registry.go+21 −0
diff --git a/docs/superpowers/plans/2026-08-28-instance-moves.md b/docs/superpowers/plans/2026-08-28-instance-moves.md| new file mode 100644 |
| index 000000000..3da28981a |
| --- /dev/null |
| +++ b/docs/superpowers/plans/2026-08-28-instance-moves.md |
| @@ -0,0 +1,109 @@ |
| +# Instance Moves Implementation Plan |
| + |
| +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. |
| + |
| +**Goal:** Two-phase, data-following instance moves between regions (issue #331): prepare mirrors the instance's replica prefix into the target region's staging area; cut stops, final-syncs, purges, and flips the record's region. |
| + |
| +**Architecture:** The record carries `move:{to,state}`; the console writes requests (+ serial bump), the SOURCE box's move worker does everything else in delta rounds; the target box needs only a staging fallback in the activator's Restore. Never two writers: the region flip is the source's last act. |
| + |
| +**Tech Stack:** Go; S3 server-side CopyObject via aws-sdk-go-v2 (already a dependency through litestream/lss3); the repo's DirStore + fake patterns for tests. |
| + |
| +**Spec:** `docs/superpowers/specs/2026-08-28-instance-move-design.md` (as amended: prepare copies the REPLICA prefix, never tails the DB twice). |
| + |
| +## Global Constraints |
| + |
| +- Wire-shape rule: every field additive + omitempty; pre-change records/edges files parse unchanged. |
| +- Move verbs accept ONLY exec-backed `kind:"instance"` (empty kind = instance) routes; fleet placements (`fleet/<name>`) refused. |
| +- Target region must be in the armed-placement set (`armedPlacements`, #329) — no dormant path for moves. |
| +- Sibling-bucket IAM stays `moves/*`-scoped: Put/Delete on objects, List with prefix condition. Never `instances/*`, never Get. |
| +- Cut ordering is sacred: route Status "moving" → child down → final mirror round → local purge → CAS flip+clear. Idempotent on resume. |
| +- `TMPDIR=/tmp go test ./...` green before any commit claims done. |
| + |
| +--- |
| + |
| +### Task 1: Record move shape + CAS transitions (internal/provision) |
| + |
| +**Files:** Modify `internal/provision/provision.go`; test `internal/provision/provision_test.go`. |
| + |
| +**Produces:** |
| +- `type MoveState struct { To string `json:"to"`; State string `json:"state"`; RequestedAt time.Time `json:"requested_at"`; ReadyAt *time.Time `json:"ready_at,omitempty"` }` |
| +- `Record.Move *MoveState `json:"move,omitempty"`` |
| +- `const MovePreparing/MoveReady/MoveCutting = "preparing"/"ready"/"cutting"` |
| +- `func SetRecordMove(ctx, s, acct, app, host string, mut func(*Record) error) error` — one narrow read-CAS-write with one retry, `SetRecordRegion`'s exact shape, generalized: every transition (request, ready-promotion, abort-clear, cut-mark, flip+clear) is a `mut` that validates the current state and returns a typed error on an illegal transition (`ErrMoveConflict`). |
| + |
| +**Steps:** |
| +- [ ] Failing tests: request-on-clean sets preparing; request-on-inflight → ErrMoveConflict; ready only from preparing; cut only from ready; abort clears preparing/ready, refuses cutting; flip sets Region=To and clears Move in ONE write; lost-CAS retried once; pre-move JSON (no `move` key) round-trips byte-identical. |
| +- [ ] Implement; `TMPDIR=/tmp go test ./internal/provision`. |
| +- [ ] Commit. |
| + |
| +### Task 2: Edges files carry data-bucket coordinates (internal/host + console reader) |
| + |
| +**Files:** Modify `internal/host/edgesfile.go` (fields `DataBucket string `json:"data_bucket,omitempty"``, `DataBucketRegion string `json:"data_bucket_region,omitempty"`` on `edgesFile`, sourced in `edgesFileLoop` from `CARLOS_HIBERNATE_BUCKET`/`CARLOS_HIBERNATE_REGION`, two new `edgesFilePass` params like `placement`); `internal/console/instancesplacement.go` (same fields on `edgesFilePlacementView`). Tests: `internal/host/edgesfile_test.go` raw-wire body with/without. |
| + |
| +- [ ] Failing raw-wire test (set → present between placement and generation; unset → absent) → implement → test → commit. |
| + |
| +### Task 3: Console move verbs |
| + |
| +**Files:** Create `internal/console/apiinstancemove.go`; modify `internal/console/server.go` (routes `POST .../instances/{host}/move`, `POST .../instances/{host}/move/cut`, `DELETE .../instances/{host}/move`); test `internal/console/apiinstancemove_test.go`. |
| + |
| +**Consumes:** Task 1 helpers; `armedPlacements` (#329); `bumpSerial` (domainattach.go's helper — reuse, don't copy); `instanceGate`. |
| + |
| +**Behaviour matrix (each a test):** 202 + record shows preparing on happy path; 400 unarmed target naming armed set (armed==false is ALSO a 400 for moves — "this deployment has no armed regions"); 400 to==current region; 400 non-exec kind (static/service/addr) and fleet placement; 409 move in flight; cut: 409 unless ready; abort: 200 clears preparing/ready, 409 cutting; every successful write bumps control/serial; 404 matrix byte-identical to the instance GET's. |
| + |
| +- [ ] Failing tests → implement → test → commit. |
| + |
| +### Task 4: Prefix mirror (internal/lease/mirror.go) |
| + |
| +**Files:** Create `internal/lease/mirror.go`, `internal/lease/mirror_test.go`. |
| + |
| +**Produces:** |
| +- `type MirrorStore interface { List(ctx, bucket, prefix string) (map[string]string, error) /* key→etag */; Copy(ctx, srcBucket, srcKey, dstBucket, dstKey string) error; Delete(ctx, bucket, key string) error }` |
| +- `func NewS3Mirror(region, dstRegion string) MirrorStore` — aws-sdk-go-v2 client(s); CopyObject with URL-encoded CopySource; dst client in the sibling bucket's region. |
| +- `func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstBucket, dstPrefix string) (changed int, err error)` — list both, copy new/etag-mismatched, delete dst keys absent at src; returns changed==0 ⇒ caught up. |
| + |
| +- [ ] Failing tests against an in-memory fake MirrorStore: initial full copy; delta only; deletion mirrored; changed==0 when identical; error propagation → implement → test → commit. (NewS3Mirror itself is exercised live in Task 8's rehearsal, not unit-tested.) |
| + |
| +### Task 5: The source-box move worker (internal/host/movespass.go) |
| + |
| +**Files:** Create `internal/host/movespass.go`, `internal/host/movespass_test.go`. |
| + |
| +**Consumes:** Task 1 transitions, Task 4 `MirrorPrefix`, `readEdgesFiles`-equivalent host-side parse of `control/edges/` for target coordinates (host has its own edges reader in steering_loop.go — `loadEdgesFileBodies` + a small parse), `registry.SetStatus` (verify exact setter name at build time; the registry has route mutators), activator quiesce via the keep-set (Task 6). |
| + |
| +**Produces:** `func movesPass(ctx context.Context, s release.Store, reg *registry.Registry, cfg moveCfg) error` called from the agent loop (Task 7); `moveCfg{Region, OwnBucket, OwnBucketRegion, Prefix string; Mirror func(dstRegion string) lease.MirrorStore; DataDir string; Quiesced func(host string) bool}`. |
| + |
| +**Semantics (each pinned by a table-driven test with DirStore + fake mirror):** |
| +- Scans records in this box's scope with `Move != nil`. |
| +- preparing: resolve target coords from edges files (placement match; missing/ambiguous → log, skip — NOT a state change); one `MirrorPrefix` round `instances/<host>/ → moves/<host>/`; if changed==0 promote to ready (CAS). Abort seen (Move cleared by console) ⇒ nothing to do — rounds simply stop. |
| +- ready: keep mirroring each pass (writes after ready still flow). |
| +- cutting, in order: set route Status "moving" (edge stops serving; RetireExcept stops the child); if child still up (Quiesced false) → return, next pass retries; final MirrorPrefix round; delete local `<DataDir>/<host>.db{,-wal,-shm}` + litestream meta dirs (reuse the exact list from `ExecBackend.Restore`'s cleanup — extract a shared helper in Task 6); CAS flip Region=To + clear Move; route retirement left to the existing reconcile pass. |
| +- Resume idempotence: every cutting step re-runnable; a flip that lost the CAS because it already happened is terminal success. |
| +- The pass NEVER acts on a record whose Region ≠ cfg.Region (post-flip: source stops seeing it; target never runs movespass for it). |
| + |
| +- [ ] Failing tests (state table: prepare-not-caught-up no-op, prepare-caught-up promotes, ready keeps mirroring, cut full sequence ordering asserted via fake-mirror/op log, cut waits for quiesce, resume-from-cutting, missing target coords skip) → implement → test → commit. |
| + |
| +### Task 6: Activator: staging-fallback restore + shared purge + quiesce visibility |
| + |
| +**Files:** Modify `internal/activator/backend_exec.go`, `internal/activator/activator.go`; test in `internal/activator/backend_exec_test.go`. |
| + |
| +**Produces:** |
| +- `func PurgeLocalDB(dataDir, host, dbPath string)` — the one list of local artifacts (db, -wal, -shm, both litestream meta dirs), used by Restore's pre-restore cleanup AND Task 5's cut. |
| +- `ExecBackend.Restore` fallback: own `instances/<host>/` has no replica AND `moves/<host>/` exists in own bucket → server-side copy `moves/→instances/` (via a `Mirror lease.MirrorStore` field on ExecBackend, nil = no fallback), then restore as normal; delete `moves/<host>/` only after `CheckDB` passes; corrupt staged replica quarantines exactly like a corrupt own one and leaves staging in place (evidence). |
| +- `Service.Idle(host) bool` (or reuse an existing accessor if one exists — check `entries` map) for Task 5's `Quiesced`. |
| + |
| +- [ ] Failing tests (fallback copy+restore+cleanup; own-replica-beats-staging precedence; corrupt staging quarantines, staging kept; nil Mirror = old behaviour) → implement → test → commit. |
| + |
| +### Task 7: Agent wiring |
| + |
| +**Files:** Modify `internal/host/agent.go` (arm movesPass on the same trigger as the reconcile pass, gated like every duty: only when deployment bucket + CARLOS_REGION set — a box with no placement code can still be a move SOURCE? NO: a record with `region:""` lives on the home box which has no code; moves FROM the home box need cfg.Region==record.Region — "" == "" matches, so arm whenever the deployment bucket exists and pass cfg.Region verbatim, empty included); `ExecBackend.Mirror = lease.NewS3Mirror(...)` when hibernate bucket configured. |
| + |
| +- [ ] Wire, `go vet ./...`, full `TMPDIR=/tmp go test ./...`, commit. |
| + |
| +### Task 8: IAM module + rollout collateral |
| + |
| +**Files:** Modify `infra/modules/region-host/variables.tf` + `main.tf` (`variable "sibling_staging" { type = list(object({bucket_arn=string})) , default = [] }` → one policy statement: PutObject+DeleteObject on `${arn}/moves/*`, ListBucket on `${arn}` with `s3:prefix` condition `moves/*`); same for `region-remote-host`. README env/di contract notes (CARLOS_REGION on the home box optional, movesPass duty, staging prefix). |
| + |
| +- [ ] `tofu fmt` + `tofu validate` in the module dir (no apply — applies are per-deployment, tito's are tito-side per the 2026-08-28 ruling); commit. |
| + |
| +### Task 9: Gate + PR |
| + |
| +- [ ] Full suite + vet; self-review diff; PR referencing #331 with the amendment called out; live-rehearsal plan (flagship: create a canary instance in `us`, move it `us→uk`, prove serve + data survival, move back) listed in the PR as the deploy-time verification — NOT run until the fleet carries the binary and the IAM grant is applied. |
diff --git a/docs/superpowers/specs/2026-08-28-instance-move-design.md b/docs/superpowers/specs/2026-08-28-instance-move-design.md| index f283f8a97..571c75eaf 100644 |
| --- a/docs/superpowers/specs/2026-08-28-instance-move-design.md |
| +++ b/docs/superpowers/specs/2026-08-28-instance-move-design.md |
| @@ -36,7 +36,7 @@ move is in flight: |
| ```json |
| "move": { |
| "to": "uk", |
| - "state": "preparing" | "ready" | "cutting", |
| + "state": "preparing" | "ready" | "cutting" | "aborting", |
| "requested_at": "<RFC3339>", |
| "ready_at": "<RFC3339, when state >= ready>" |
| } |
| @@ -67,48 +67,66 @@ instance itself): |
| calls this only after freezing writes; the platform does not verify |
| the freeze (it cannot — the freeze is app-level), it documents it. |
| - `DELETE .../instances/{host}/move` — abort, legal in `preparing` or |
| - `ready` (clears `move`; the source drops its second replica; staged |
| - bytes in the target bucket are garbage that the next prepare |
| - overwrites and the target's post-restore cleanup deletes). Refused |
| - 409 in `cutting`: past that point the move completes forward. |
| + `ready`. TWO-PHASE (review round 1): the door marks `aborting`; the |
| + SOURCE box deletes the staged copy from the target bucket and then |
| + clears `move` — a bare clear would leave staging behind to win a |
| + later landing on the target (staging-wins below). Refused 409 in |
| + `cutting` (completes forward) and in `aborting` (already under way). |
| - State is read, not pushed: the existing single-record GET already |
| returns the record verbatim, so `move.state` is pollable today with |
| zero new read surface. |
| +- Every move write bumps `control/serial` (the config fast-convergence |
| + nudge the env door already uses), so the source box notices in |
| + seconds, not on the hourly floor. Between states the source's own |
| + move worker self-paces its delta rounds. |
| ## What the source box does |
| The reconcile/activator pass on the box whose `CARLOS_REGION` matches |
| `region` (the source — the record still points at it): |
| -1. **preparing**: alongside its normal own-bucket replication, the |
| - activator opens a SECOND litestream replica of the instance DB to |
| - `s3://<target-data-bucket>/moves/<host>/` — the lease package's |
| - `ReplicateWithLease` pointed at the sibling bucket's staging prefix. |
| +1. **preparing** (AMENDED during planning, 2026-08-28: the draft said |
| + "a second litestream replica of the live DB" — two litestream |
| + processes cannot tail one WAL, both want to own checkpointing. The |
| + fallback IS the design, and it is better): the move copies the |
| + REPLICA, never the database. The activator already keeps |
| + `instances/<host>/` in the source's own bucket continuously synced |
| + (running child) or final-checkpointed (parked child) — the move |
| + worker mirrors that prefix into |
| + `s3://<target-data-bucket>/moves/<host>/` with server-side |
| + CopyObject (its IAM: read own bucket, push sibling staging), in |
| + delta rounds every ~30s — List both sides, copy new/changed keys, |
| + delete keys the source pruned (a true mirror, so litestream |
| + retention holds in staging too). Uniform for running and parked |
| + instances; no wake, no litestream internals touched. |
| Target bucket name and AWS region are derived, not configured |
| per-move: each box publishes its data bucket coordinates as two new |
| additive fields in its edges file (`data_bucket`, `data_bucket_region` |
| — same additive rule as `placement`), so the source reads the |
| target's coordinates from the same file set everything else reads. |
| - A wake is forced if the instance is asleep? NO — a parked instance's |
| - state is already fully in the source bucket; for a parked instance |
| - the "second replica" is a bucket-to-staging copy performed by the |
| - source box (its IAM can read its own bucket and push to the sibling's |
| - staging), and prepare completes without ever waking the child. |
| -2. **ready promotion**: once the staging replica reports caught up |
| - (litestream generation synced; for the parked case, copy complete), |
| - the source CASes `move.state: "preparing" → "ready"`, stamping |
| - `ready_at`. The second replica keeps streaming — ready is a level, |
| - not an edge; writes after ready keep flowing to staging. |
| -3. **cutting**: on seeing `state:"cutting"`, the source (in order): |
| - final `SyncAndWait` on the staging replica → stop the child (normal |
| - stopping path, which also checkpoints to the source's OWN bucket — |
| - the abort-recovery copy) → close both replicators → **delete the |
| - local DB, WAL/SHM, and litestream meta dirs** (the warm-cache purge; |
| - without it a later move back would resurrect a stale local file — the |
| - activator's warm-cache trust is exactly right for crashes and exactly |
| - wrong here) → one CAS that sets `region: <to>` and clears `move` → |
| - remove its own route row on its next pass (the record has left its |
| - scope, the existing retirement path does this unprompted). |
| +2. **ready promotion**: once a delta round completes with zero |
| + differences against a fresh listing, the source CASes |
| + `move.state: "preparing" → "ready"`, stamping `ready_at`. Rounds |
| + keep running — ready is a level, not an edge; writes after ready |
| + keep flowing into staging on the next round. |
| +3. **cutting** (order amended, review round 1): on seeing |
| + `state:"cutting"`, the source (in order): set the route's Status to |
| + "moving" (a non-ready status: the edge |
| + stops serving it, the edges file drops it, and the activator's |
| + RetireExcept sweep stops the child through the ordinary stop path — |
| + which final-checkpoints to the source's OWN bucket — but SWALLOWS a |
| + checkpoint failure, safe only while the warm cache stood behind it) |
| + → wait for the child to be down → **re-checkpoint the local DB with |
| + the lease and wait** (proving the own replica carries the tail — |
| + review round 1, the failed-final-checkpoint data-loss hole) → one |
| + final delta round into staging (now byte-identical to the own replica) → |
| + **delete the local DB, WAL/SHM, and litestream meta dirs** (the |
| + warm-cache purge; without it a later move back would resurrect a |
| + stale local file — the activator's warm-cache trust is exactly right |
| + for crashes and exactly wrong here) → one CAS that sets |
| + `region: <to>` and clears `move` → remove its own route row on its |
| + next pass (the record has left its scope, the existing retirement |
| + path does this unprompted). |
| If the box dies mid-cut, the record still says `cutting` with the old |
| region: the source resumes the sequence on restart (every step is |
| idempotent — sync again, stop a stopped child, delete absent files, |
| @@ -136,12 +154,11 @@ recovery door that already exists). |
| List, no Delete, never `instances/*`. Residency-at-rest holds |
| (cross-region reads remain impossible at IAM); the write blast radius |
| is confined to staging keys no box ever trusts as live state without |
| -its own restore-time `CheckDB`. Litestream against a push-only prefix: |
| -the lease/generation dance needs List on its own prefix — checked |
| -during build; if litestream cannot run push-only, prepare falls back to |
| -periodic own-checkpoint + copy-into-staging (same convergence, chunkier |
| -cadence), and the spec's contract (`ready` = staged copy caught up) |
| -does not change. |
| +its own restore-time `CheckDB`. The mirror needs three actions on the |
| +sibling, all `moves/*`-scoped: `s3:PutObject` and `s3:DeleteObject` on |
| +the object ARNs, and `s3:ListBucket` on the bucket ARN with an |
| +`s3:prefix` condition of `moves/*` — still no reads of sibling data, |
| +still nothing outside staging. |
| ## Instance env |
| @@ -159,6 +176,15 @@ titogo's boot-clears-freeze rule needs exactly this. |
| (400 naming the kind) — the only class whose state the platform owns. |
| - Fleet placements (`fleet/<name>`): refused; fleets keep D6 |
| state-does-not-follow semantics on purpose. |
| +- Pooled records (`pool != ""`): refused in v1 — their data lives on |
| + the pool's box, not the platform region box that would drive the |
| + move. Non-hibernating instances (`hibernate:false`): refused — they |
| + run outside the activator, so nothing manages (or could quiesce) |
| + their data. |
| +- The worker acts only on routes its OWN registry serves: region |
| + equality alone would let a standby or second same-region box drive a |
| + cut whose child runs on its sibling (per-box registries make its |
| + Quiesced trivially true). |
| - Automatic freeze verification; region-aware unit economics; the |
| member-facing residence door (titogo builds on these verbs). |
diff --git a/infra/modules/region-host/main.tf b/infra/modules/region-host/main.tf| index ae5afb52f..deb71e96f 100644 |
| --- a/infra/modules/region-host/main.tf |
| +++ b/infra/modules/region-host/main.tf |
| @@ -78,7 +78,12 @@ resource "aws_iam_role_policy" "host_buckets" { |
| role = aws_iam_role.host.id |
| policy = jsonencode({ |
| Version = "2012-10-17" |
| - Statement = [ |
| + # The two sibling-staging statement groups (instance-move spec, issue |
| + # #331) are appended only when sibling_staging_arns is non-empty: |
| + # push-only into moves/* — a box can stage a residence move TOWARD a |
| + # sibling region but can never read that region's data or touch its |
| + # live instances/* replica lineage. |
| + Statement = concat([ |
| { |
| Effect = "Allow" |
| Action = ["s3:GetObject", "s3:PutObject", "s3:DeleteObject"] |
| @@ -131,7 +136,19 @@ resource "aws_iam_role_policy" "host_buckets" { |
| ] |
| Resource = "*" |
| }, |
| - ] |
| + ], |
| + [for arn in var.sibling_staging_arns : { |
| + Effect = "Allow" |
| + Action = ["s3:PutObject", "s3:DeleteObject"] |
| + Resource = "${arn}/moves/*" |
| + }], |
| + [for arn in var.sibling_staging_arns : { |
| + Effect = "Allow" |
| + Action = ["s3:ListBucket"] |
| + Resource = arn |
| + Condition = { StringLike = { "s3:prefix" = "moves/*" } } |
| + }] |
| + ) |
| }) |
| } |
diff --git a/infra/modules/region-host/variables.tf b/infra/modules/region-host/variables.tf| index 3f0a898c0..b508af535 100644 |
| --- a/infra/modules/region-host/variables.tf |
| +++ b/infra/modules/region-host/variables.tf |
| @@ -114,3 +114,9 @@ variable "alerts_email" { |
| type = string |
| default = "" |
| } |
| + |
| +variable "sibling_staging_arns" { |
| + description = "Sibling regions' data-bucket ARNs this host may STAGE residence-moves into (instance-move spec, issue #331). The grant is moves/*-scoped push-only: Put/Delete on the staging objects and a prefix-conditioned ListBucket — never instances/*, never GetObject, so residency-at-rest holds while a move's delta rounds run. Empty (the default): no cross-bucket statement at all." |
| + type = list(string) |
| + default = [] |
| +} |
diff --git a/infra/modules/region-remote-host/main.tf b/infra/modules/region-remote-host/main.tf| index ce7d0ecf5..d18648e76 100644 |
| --- a/infra/modules/region-remote-host/main.tf |
| +++ b/infra/modules/region-remote-host/main.tf |
| @@ -74,7 +74,11 @@ resource "aws_iam_user_policy" "host" { |
| user = aws_iam_user.host.name |
| policy = jsonencode({ |
| Version = "2012-10-17" |
| - Statement = [ |
| + # Sibling staging (instance-move spec, issue #331) — same push-only |
| + # moves/* shape as region-host's host_buckets policy; keep the two in |
| + # step (this module's README already carries that rule for the S3 |
| + # action list). |
| + Statement = concat([ |
| { |
| Sid = "OwnRegionDataObjects" |
| Effect = "Allow" |
| @@ -87,7 +91,19 @@ resource "aws_iam_user_policy" "host" { |
| Action = ["s3:ListBucket", "s3:GetBucketLocation"] |
| Resource = [aws_s3_bucket.data.arn] |
| }, |
| - ] |
| + ], |
| + [for arn in var.sibling_staging_arns : { |
| + Effect = "Allow" |
| + Action = ["s3:PutObject", "s3:DeleteObject"] |
| + Resource = "${arn}/moves/*" |
| + }], |
| + [for arn in var.sibling_staging_arns : { |
| + Effect = "Allow" |
| + Action = ["s3:ListBucket"] |
| + Resource = arn |
| + Condition = { StringLike = { "s3:prefix" = "moves/*" } } |
| + }] |
| + ) |
| }) |
| } |
diff --git a/infra/modules/region-remote-host/variables.tf b/infra/modules/region-remote-host/variables.tf| index 2096ca518..9fe68ecd6 100644 |
| --- a/infra/modules/region-remote-host/variables.tf |
| +++ b/infra/modules/region-remote-host/variables.tf |
| @@ -13,3 +13,9 @@ variable "deployment_bucket_arn" { |
| type = string |
| default = "" |
| } |
| + |
| +variable "sibling_staging_arns" { |
| + description = "Sibling regions' data-bucket ARNs this host may STAGE residence-moves into (instance-move spec, issue #331). moves/*-scoped push-only, exactly region-host's statement shape: Put/Delete on staging objects + prefix-conditioned ListBucket; never instances/*, never GetObject. Empty (the default): no cross-bucket statement." |
| + type = list(string) |
| + default = [] |
| +} |
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index 2ec94c275..609ccf743 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -1080,6 +1080,18 @@ func (a *Service) RetireExcept(keep map[string]bool) { |
| } |
| } |
| +// Quiesced answers whether a host currently has NO live activator entry — |
| +// no child running, none waking, none mid-stop. The residence-move cut |
| +// (internal/host's move worker) gates its final mirror round on this: the |
| +// route's Status change retires the entry through RetireExcept, and only a |
| +// quiet host's replica prefix is byte-final. |
| +func (a *Service) Quiesced(host string) bool { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + _, live := a.entries[host] |
| + return !live |
| +} |
| + |
| // Shutdown checkpoints and stops every child — the agent going down takes its |
| // instances with it (they're its children; leaving them orphaned would strand |
| // the leases until TTL expiry and leave the sockets bound against the next |
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go| index a8fac9f93..995870cbf 100644 |
| --- a/internal/activator/backend_exec.go |
| +++ b/internal/activator/backend_exec.go |
| @@ -60,6 +60,14 @@ type ExecBackend struct { |
| // but a carlos app is a separate binary from the release store — there is |
| // nothing sensible to invent here, and the adopt pass owns the symlinks. |
| Materialize func(ctx context.Context, host string) error |
| + // Mirror, when non-nil, arms Restore's staging fallback (issue #331): |
| + // a host with no replica under the ordinary instances/ prefix may have |
| + // one under moves/<host>/ — a residence move's staged copy, pushed by |
| + // the SOURCE box. Restore then copies it into place (own bucket, own |
| + // IAM), restores as normal, and deletes the staging copy once CheckDB |
| + // vouches for what it restored. Nil = the fallback does not exist, |
| + // byte-identical to the pre-move behaviour. |
| + Mirror lease.MirrorStore |
| // Region is the box's CARLOS placement code (CARLOS_REGION), handed to |
| // every exec child as its own CARLOS_REGION (issue #329, titogo's |
| // residence design: the app reads where it lives — e.g. to clear a move |
| @@ -226,11 +234,27 @@ func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, err |
| // its MetaDirSuffix — spelled out because this package never imports |
| // litestream) as any box-wide backup replicator would use, and the lease |
| // package's activator-scoped ".activator-<db>-litestream". |
| - for _, p := range []string{dbPath + "-wal", dbPath + "-shm"} { |
| - _ = os.Remove(p) |
| + PurgeLocalDBArtifacts(dbPath, false) |
| + |
| + // The staging fallback (issue #331): when the ordinary prefix is empty |
| + // and a move staged a replica here, promote it — server-side, own |
| + // bucket — and restore from the promoted copy. Precedence is the |
| + // ordinary prefix ALWAYS: a host that re-lands here after moving away |
| + // and back must restore its own lineage, never regress to an old |
| + // staged copy (the source's cut deleted the own prefix it left behind, |
| + // so a non-empty own prefix is by construction the newer story). |
| + promoted := false |
| + if b.Mirror != nil { |
| + var err error |
| + if promoted, err = b.promoteStagedReplica(ctx, host); err != nil { |
| + // HARD failure, same rule as restoreReplica's own error paths |
| + // (review round 1): "continuing without it" here would fall |
| + // through to a FRESH boot + seed, and once that empty database |
| + // replicates, the staged data is shadowed forever. A failed |
| + // wake retries; a seeded ghost does not. |
| + return false, fmt.Errorf("staging fallback for %s: %w", host, err) |
| + } |
| } |
| - _ = os.RemoveAll(dbPath + "-litestream") |
| - _ = os.RemoveAll(filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream")) |
| restored, err := restoreReplica(ctx, dbPath, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region) |
| if err != nil { |
| @@ -270,6 +294,14 @@ func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, err |
| return false, fmt.Errorf("restored replica for %s is corrupt: %w", host, err) |
| } |
| log.Printf("activator: restored %s from s3://%s/%s", host, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host)) |
| + // A clean restore of a copy promoted THIS call is the moment the |
| + // staging has served its purpose; a corrupt one (above) |
| + // deliberately LEFT it in place as evidence, and staging that was |
| + // never consulted (an abort's leftovers under a live own lineage) |
| + // is not this restore's to touch. |
| + if promoted { |
| + b.dropStagedReplica(ctx, host) |
| + } |
| } else { |
| // Never replicated: a brand-new instance boots fresh and creates its |
| // own DB — the register-only create path (provisioning as a metadata |
| @@ -999,3 +1031,71 @@ func (p *childProc) Stop(ctx context.Context) error { |
| return fmt.Errorf("instance did not exit on SIGTERM; killed") |
| } |
| } |
| + |
| +// movesPrefix is where a residence move stages a replica in the TARGET's |
| +// data bucket (instance-move spec): moves/<host>/ beside the ordinary |
| +// instances/ prefix. A free function so the host's move worker and this |
| +// package spell it identically. |
| +func MovesPrefix(host string) string { return "moves/" + host + "/" } |
| + |
| +// PurgeLocalDBArtifacts removes the local litestream footprint next to a |
| +// DB — WAL, SHM and both meta dirs — and, when full is true, the DB file |
| +// itself. Restore's pre-restore cleanup uses the partial form (the DB's |
| +// absence is what got us here); a residence move's cut uses the full form: |
| +// the activator's warm-cache trust is exactly right for crashes and exactly |
| +// wrong after a cut, where a stale local file would resurrect a moved-away |
| +// instance's history. |
| +func PurgeLocalDBArtifacts(dbPath string, full bool) { |
| + for _, p := range []string{dbPath + "-wal", dbPath + "-shm"} { |
| + _ = os.Remove(p) |
| + } |
| + _ = os.RemoveAll(dbPath + "-litestream") |
| + _ = os.RemoveAll(filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream")) |
| + if full { |
| + _ = os.Remove(dbPath) |
| + } |
| +} |
| + |
| +// promoteStagedReplica copies moves/<host>/ over instances/<host>/ in this |
| +// box's OWN bucket when — and only when — the ordinary prefix is empty. A |
| +// no-op (nil error) when there is nothing staged. |
| +func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (bool, error) { |
| + // STAGING WINS when present (review round 1). A staged replica exists |
| + // only while an inbound move is landing or has just landed: it was |
| + // mirrored from the then-live data at cut, so it is by construction |
| + // newer than any lineage this box kept from a PREVIOUS residence — the |
| + // boomerang case (ie→uk→ie) that "own lineage never loses" restored |
| + // months-stale data from. An abandoned move's staging can't linger to |
| + // poison this: abort is two-phase, and the source deletes staging |
| + // before the move clears (MoveAborting). |
| + // |
| + // The trailing slash on the guard listing matters: hibernatePrefix has |
| + // none (path.Join), and a slashless List would let app.example.com's |
| + // keys answer for app.example.co. |
| + staged, err := b.Mirror.List(ctx, b.Cfg.Bucket, MovesPrefix(host)) |
| + if err != nil { |
| + return false, err |
| + } |
| + if len(staged) == 0 { |
| + return false, nil |
| + } |
| + log.Printf("activator: %s: promoting %d staged object(s) from %s (residence move landing)", host, len(staged), MovesPrefix(host)) |
| + if _, err := lease.MirrorPrefix(ctx, b.Mirror, b.Cfg.Bucket, MovesPrefix(host), b.Cfg.Bucket, b.Cfg.hibernatePrefix(host)); err != nil { |
| + return false, err |
| + } |
| + return true, nil |
| +} |
| + |
| +// dropStagedReplica deletes moves/<host>/ after a restore CheckDB vouched |
| +// for. Best-effort: a leftover staging copy is storage, not a correctness |
| +// problem — promoteStagedReplica never consults it again once the ordinary |
| +// prefix exists. |
| +func (b *ExecBackend) dropStagedReplica(ctx context.Context, host string) { |
| + staged, err := b.Mirror.List(ctx, b.Cfg.Bucket, MovesPrefix(host)) |
| + if err != nil { |
| + return |
| + } |
| + for k := range staged { |
| + _ = b.Mirror.Delete(ctx, b.Cfg.Bucket, k) |
| + } |
| +} |
diff --git a/internal/activator/backend_exec_test.go b/internal/activator/backend_exec_test.go| index 7e3fb3a8a..84f291c12 100644 |
| --- a/internal/activator/backend_exec_test.go |
| +++ b/internal/activator/backend_exec_test.go |
| @@ -2,6 +2,7 @@ package activator |
| import ( |
| "context" |
| + "database/sql" |
| "os" |
| "os/exec" |
| "path/filepath" |
| @@ -9,6 +10,8 @@ import ( |
| "testing" |
| "time" |
| + _ "modernc.org/sqlite" |
| + |
| "github.com/carlosframework/platform/internal/cgroup" |
| "github.com/carlosframework/platform/internal/registry" |
| ) |
| @@ -103,3 +106,142 @@ func TestExecBackendStartChildRegion(t *testing.T) { |
| }) |
| } |
| } |
| + |
| +// The staging fallback (issue #331): a host whose ordinary prefix is empty |
| +// restores from a move's staged copy — promoted server-side, dropped after |
| +// a clean restore — and the ordinary prefix ALWAYS wins when it exists. |
| +func TestRestoreStagingFallback(t *testing.T) { |
| + ctx := context.Background() |
| + b := testExecBackend(t) |
| + f := newFakeMirrorStore() |
| + b.Mirror = f |
| + |
| + host := "moved.example.com" |
| + rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "i.sock")} |
| + |
| + // restoreReplica is stubbed for the WHOLE test (the real one dials S3): |
| + // it restores — materializing a genuinely valid SQLite file, so CheckDB |
| + // passes — exactly when the fake bucket holds objects under the prefix |
| + // it was pointed at, which is the promotion's observable effect. |
| + var restoredFrom []string |
| + oldRestore := restoreReplica |
| + restoreReplica = func(_ context.Context, dbPath, bucket, prefix, region string) (bool, error) { |
| + objs, _ := f.List(context.Background(), bucket, prefix) |
| + if len(objs) == 0 { |
| + return false, nil |
| + } |
| + restoredFrom = append(restoredFrom, bucket+"/"+prefix) |
| + return true, writeMinimalSQLite(dbPath) |
| + } |
| + defer func() { restoreReplica = oldRestore }() |
| + |
| + // Nothing staged, nothing own: plain fresh boot (restored=false→fresh=true). |
| + fresh, err := b.Restore(ctx, rt) |
| + if err != nil || !fresh { |
| + t.Fatalf("empty everything: fresh=%v err=%v, want fresh boot", fresh, err) |
| + } |
| + |
| + // Staged copy present, own prefix empty: promoted then restored. |
| + f.put(b.Cfg.Bucket, "moves/"+host+"/generations/a/snapshot", "e1") |
| + _ = os.Remove(filepath.Join(b.DataDir, host+".db")) |
| + fresh, err = b.Restore(ctx, rt) |
| + if err != nil { |
| + t.Fatalf("staged restore: %v", err) |
| + } |
| + if fresh { |
| + t.Fatal("staged restore reported a fresh boot") |
| + } |
| + if len(restoredFrom) != 1 || restoredFrom[0] != b.Cfg.Bucket+"/"+b.Cfg.hibernatePrefix(host) { |
| + t.Fatalf("restored from %v, want the promoted ordinary prefix %s", restoredFrom, b.Cfg.hibernatePrefix(host)) |
| + } |
| + own, _ := f.List(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host)) |
| + if len(own) != 1 { |
| + t.Fatalf("promotion left own prefix = %v", own) |
| + } |
| + staged, _ := f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/") |
| + if len(staged) != 0 { |
| + t.Fatalf("staging not dropped after a clean restore: %v", staged) |
| + } |
| + |
| + // STAGING WINS (review round 1): a fresh staged copy beside a stale |
| + // own lineage is the boomerang landing — the staged bytes replace the |
| + // own prefix and are what gets restored. |
| + f.put(b.Cfg.Bucket, "moves/"+host+"/generations/b/newer-snapshot", "e9") |
| + _ = os.Remove(filepath.Join(b.DataDir, host+".db")) |
| + restoredFrom = nil |
| + if _, err := b.Restore(ctx, rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + own, _ = f.List(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host)) |
| + if len(own) != 1 { |
| + t.Fatalf("own prefix after boomerang promote = %v, want exactly the staged copy", own) |
| + } |
| + for k := range own { |
| + if !strings.Contains(k, "newer-snapshot") { |
| + t.Fatalf("own prefix kept the stale lineage: %v", own) |
| + } |
| + } |
| + staged, _ = f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/") |
| + if len(staged) != 0 { |
| + t.Fatalf("staging not dropped after the boomerang restore: %v", staged) |
| + } |
| + |
| + // A staging-fallback ERROR fails the wake — never a fresh boot + seed |
| + // that would shadow the staged data forever. |
| + f.put(b.Cfg.Bucket, "moves/"+host+"/generations/c/snapshot", "eA") |
| + f.failList = true |
| + _ = os.Remove(filepath.Join(b.DataDir, host+".db")) |
| + if _, err := b.Restore(ctx, rt); err == nil { |
| + t.Fatal("List failure in the staging fallback did not fail the restore") |
| + } |
| + f.failList = false |
| +} |
| + |
| +// fakeMirrorStore is the test double for lease.MirrorStore, bucket→key→etag. |
| +type fakeMirrorStore struct { |
| + buckets map[string]map[string]string |
| + failList bool |
| +} |
| + |
| +func newFakeMirrorStore() *fakeMirrorStore { |
| + return &fakeMirrorStore{buckets: map[string]map[string]string{}} |
| +} |
| +func (f *fakeMirrorStore) put(bucket, key, etag string) { |
| + if f.buckets[bucket] == nil { |
| + f.buckets[bucket] = map[string]string{} |
| + } |
| + f.buckets[bucket][key] = etag |
| +} |
| +func (f *fakeMirrorStore) List(_ context.Context, bucket, prefix string) (map[string]string, error) { |
| + if f.failList { |
| + return nil, context.DeadlineExceeded |
| + } |
| + out := map[string]string{} |
| + for k, v := range f.buckets[bucket] { |
| + if strings.HasPrefix(k, prefix) { |
| + out[k] = v |
| + } |
| + } |
| + return out, nil |
| +} |
| +func (f *fakeMirrorStore) Copy(_ context.Context, srcBucket, srcKey, dstBucket, dstKey string) error { |
| + f.put(dstBucket, dstKey, f.buckets[srcBucket][srcKey]) |
| + return nil |
| +} |
| +func (f *fakeMirrorStore) Delete(_ context.Context, bucket, key string) error { |
| + delete(f.buckets[bucket], key) |
| + return nil |
| +} |
| + |
| +// writeMinimalSQLite creates a genuinely valid empty SQLite database so |
| +// lease.CheckDB passes on the stubbed restore's output. |
| +func writeMinimalSQLite(dbPath string) error { |
| + db, err := sql.Open("sqlite", dbPath) |
| + if err != nil { |
| + return err |
| + } |
| + if _, err := db.Exec("create table t(x)"); err != nil { |
| + return err |
| + } |
| + return db.Close() |
| +} |
diff --git a/internal/console/apiinstancemove.go b/internal/console/apiinstancemove.go| new file mode 100644 |
| index 000000000..69cf3f836 |
| --- /dev/null |
| +++ b/internal/console/apiinstancemove.go |
| @@ -0,0 +1,156 @@ |
| +package console |
| + |
| +import ( |
| + "encoding/json" |
| + "errors" |
| + "log" |
| + "net/http" |
| + "os" |
| + "slices" |
| + "strings" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/registry" |
| +) |
| + |
| +// The residence-move doors (issue #331, instance-move spec): the console |
| +// writes REQUESTS onto the record and bumps the serial; everything that |
| +// touches data happens on the source box's move worker. Three verbs: |
| +// |
| +// POST .../instances/{host}/move {"region":"uk"} → preparing |
| +// POST .../instances/{host}/move/cut → cutting |
| +// DELETE .../instances/{host}/move → aborted |
| +// |
| +// State is read through the ordinary instance GET — the record carries |
| +// `move` verbatim while one is in flight. |
| + |
| +// moveGate is the shared front half: the instance family's gate, the |
| +// byte-identical 404 for a host that names no record, and the class rail — |
| +// only an exec-backed instance route's state is the platform's to move |
| +// (spec §Out of scope), and fleet placements keep their deliberate |
| +// state-does-not-follow semantics. |
| +func (s *Server) moveGate(w http.ResponseWriter, r *http.Request) (acct *Account, app, host string, rec *provision.Record, ok bool) { |
| + _, acct, app, gok := s.instanceGate(w, r) |
| + if !gok { |
| + return nil, "", "", nil, false |
| + } |
| + host = r.PathValue("host") |
| + if !provision.ValidHostShape(host) { |
| + http.NotFound(w, r) |
| + return nil, "", "", nil, false |
| + } |
| + ctx := r.Context() |
| + rec, err := provision.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app, host) |
| + if errors.Is(err, os.ErrNotExist) { |
| + http.NotFound(w, r) |
| + return nil, "", "", nil, false |
| + } |
| + if err != nil { |
| + log.Printf("console: api instance move %s/%s/%s: %v", acct.Sqid, app, host, err) |
| + jsonError(w, http.StatusBadGateway, "could not read the instance") |
| + return nil, "", "", nil, false |
| + } |
| + if k := rec.Kind; k != "" && k != "instance" { |
| + jsonError(w, http.StatusBadRequest, "only exec-backed instances move — this route is kind "+k) |
| + return nil, "", "", nil, false |
| + } |
| + if rec.Addr != "" || len(rec.Addrs) > 0 { |
| + jsonError(w, http.StatusBadRequest, "an addr-backed route's process is not platform-managed — nothing to move") |
| + return nil, "", "", nil, false |
| + } |
| + if _, fleet := registry.FleetPlacement(rec.Region); fleet { |
| + jsonError(w, http.StatusBadRequest, "fleet-placed instances keep state-does-not-follow semantics — move is for platform regions") |
| + return nil, "", "", nil, false |
| + } |
| + if rec.Pool != "" { |
| + jsonError(w, http.StatusBadRequest, "pooled instances do not move in v1 — their data lives on the pool's own box") |
| + return nil, "", "", nil, false |
| + } |
| + if !rec.HibernateOrDefault() { |
| + jsonError(w, http.StatusBadRequest, "a non-hibernating instance runs outside the activator — nothing manages its data, nothing can move it") |
| + return nil, "", "", nil, false |
| + } |
| + return acct, app, host, rec, true |
| +} |
| + |
| +// moveTransition applies one guarded transition and maps its outcomes: |
| +// ErrMoveConflict → 409 with the transition's own sentence, success → the |
| +// fresh record + a serial bump so the source box notices in seconds. |
| +func (s *Server) moveTransition(w http.ResponseWriter, r *http.Request, acct *Account, app, host string, mut func(*provision.Record) error, okStatus int) { |
| + ctx := r.Context() |
| + err := provision.SetRecordMove(ctx, s.cfg.Releases, acct.Sqid, app, host, mut) |
| + if errors.Is(err, provision.ErrMoveConflict) { |
| + jsonError(w, http.StatusConflict, err.Error()) |
| + return |
| + } |
| + if err != nil { |
| + log.Printf("console: api instance move %s/%s/%s: %v", acct.Sqid, app, host, err) |
| + jsonError(w, http.StatusBadGateway, "could not write the move — try again") |
| + return |
| + } |
| + s.bumpSerial(ctx) |
| + rec, err := provision.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app, host) |
| + if err != nil { |
| + writeJSON(w, okStatus, map[string]any{"ok": true}) |
| + return |
| + } |
| + writeJSON(w, okStatus, rec) |
| +} |
| + |
| +func (s *Server) handleAPIInstanceMoveRequest(w http.ResponseWriter, r *http.Request) { |
| + acct, app, host, _, ok := s.moveGate(w, r) |
| + if !ok { |
| + return |
| + } |
| + var req struct { |
| + Region string `json:"region"` |
| + } |
| + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { |
| + jsonError(w, http.StatusBadRequest, "invalid JSON body") |
| + return |
| + } |
| + to := strings.TrimSpace(req.Region) |
| + if to == "" || !provision.ValidRegion(to) { |
| + jsonError(w, http.StatusBadRequest, "invalid region") |
| + return |
| + } |
| + // The armed gate is MANDATORY for moves — unlike create there is no |
| + // dormant path, because a move needs a live target box to restore on; |
| + // a deployment with no armed placements cannot be a move target at all. |
| + set, armed := armedPlacements(r.Context(), s.cfg.Releases) |
| + if !armed { |
| + jsonError(w, http.StatusBadRequest, "this deployment publishes no armed regions — nothing to move to") |
| + return |
| + } |
| + if !set[to] { |
| + codes := make([]string, 0, len(set)) |
| + for c := range set { |
| + codes = append(codes, c) |
| + } |
| + slices.Sort(codes) |
| + jsonError(w, http.StatusBadRequest, "region "+to+" is not armed in this deployment — armed: "+strings.Join(codes, ", ")) |
| + return |
| + } |
| + s.moveTransition(w, r, acct, app, host, provision.MoveRequest(to, time.Now()), http.StatusAccepted) |
| +} |
| + |
| +func (s *Server) handleAPIInstanceMoveCut(w http.ResponseWriter, r *http.Request) { |
| + acct, app, host, _, ok := s.moveGate(w, r) |
| + if !ok { |
| + return |
| + } |
| + s.moveTransition(w, r, acct, app, host, provision.MoveMarkCutting(), http.StatusAccepted) |
| +} |
| + |
| +// Abort is two-phase (review round 1): this door marks MoveAborting and the |
| +// SOURCE box deletes the staged copy before clearing the field — a bare |
| +// clear would leave staging behind to win a later landing on the target |
| +// (staging-wins). 202, like its siblings: the caller polls the record. |
| +func (s *Server) handleAPIInstanceMoveAbort(w http.ResponseWriter, r *http.Request) { |
| + acct, app, host, _, ok := s.moveGate(w, r) |
| + if !ok { |
| + return |
| + } |
| + s.moveTransition(w, r, acct, app, host, provision.MoveAbort(), http.StatusAccepted) |
| +} |
diff --git a/internal/console/apiinstancemove_test.go b/internal/console/apiinstancemove_test.go| new file mode 100644 |
| index 000000000..1aa6258ee |
| --- /dev/null |
| +++ b/internal/console/apiinstancemove_test.go |
| @@ -0,0 +1,184 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "fmt" |
| + "net/http" |
| + "strings" |
| + "testing" |
| + |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/provision" |
| +) |
| + |
| +// The residence-move doors (issue #331): the console writes guarded |
| +// transitions and serial bumps; the state machine's 400/409 matrix is the |
| +// tenant's whole interface, so every cell is pinned. |
| +func TestAPIInstanceMoveVerbs(t *testing.T) { |
| + fx, token := instanceFixture(t, "code-inst-move") |
| + m := titoManifest() |
| + m.Regions = nil |
| + m.Pool = "" // pooled records are refused at the move door — separate rail, pinned below |
| + enableInstances(t, fx, "titogo", m) |
| + u := instancesURL(fx, "bab", "titogo") |
| + put := func(key, body string) { |
| + t.Helper() |
| + if _, err := fx.rel.Put(context.Background(), key, []byte(body), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + // Create BEFORE arming: once uk-box publishes a placement, the #329 |
| + // gate would refuse region ie (not armed) — the record predates arming, |
| + // exactly like every real record on a fleet that grows regions later. |
| + if code, _, out := rawReq(t, http.MethodPost, u, token, `{"host":"acme.go.tito.io","region":"ie"}`); code != http.StatusCreated { |
| + t.Fatalf("create = %d %s", code, out) |
| + } |
| + put("control/edges/uk-box.json", `{"v":1,"box":"uk-box","eip":"1.2.3.4","placement":"uk","data_bucket":"data-uk","data_bucket_region":"eu-west-2","hosts":{}}`) |
| + moveURL := u + "/acme.go.tito.io/move" |
| + |
| + // Unarmed target → 400 naming the set. |
| + if code, _, out := rawReq(t, http.MethodPost, moveURL, token, `{"region":"se"}`); code != http.StatusBadRequest || !strings.Contains(out, "armed: uk") { |
| + t.Fatalf("move to unarmed = %d %s, want 400 naming the set", code, out) |
| + } |
| + // Same region → 409 (the transition's own conflict). |
| + if code, _, out := rawReq(t, http.MethodPost, u+"/acme.go.tito.io/move", token, `{"region":"ie"}`); code != http.StatusBadRequest && code != http.StatusConflict { |
| + t.Fatalf("move to current region = %d %s, want a refusal", code, out) |
| + } else if code == http.StatusBadRequest && !strings.Contains(out, "armed") { |
| + // "ie" is not armed either — refused at the armed gate, also fine. |
| + _ = out |
| + } |
| + // Cut before any move → 409. |
| + if code, _, out := rawReq(t, http.MethodPost, moveURL+"/cut", token, ""); code != http.StatusConflict { |
| + t.Fatalf("cut with no move = %d %s, want 409", code, out) |
| + } |
| + // Happy path: request → preparing, record carries the move. |
| + code, _, out := rawReq(t, http.MethodPost, moveURL, token, `{"region":"uk"}`) |
| + if code != http.StatusAccepted { |
| + t.Fatalf("move request = %d %s, want 202", code, out) |
| + } |
| + var rec provision.Record |
| + if err := json.Unmarshal([]byte(out), &rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + if rec.Move == nil || rec.Move.State != provision.MovePreparing || rec.Move.To != "uk" || rec.Region != "ie" { |
| + t.Fatalf("after request: region=%q move=%+v", rec.Region, rec.Move) |
| + } |
| + // Second request while in flight → 409. |
| + if code, _, out := rawReq(t, http.MethodPost, moveURL, token, `{"region":"uk"}`); code != http.StatusConflict { |
| + t.Fatalf("second move = %d %s, want 409", code, out) |
| + } |
| + // Cut from preparing → 409 (only ready cuts). |
| + if code, _, out := rawReq(t, http.MethodPost, moveURL+"/cut", token, ""); code != http.StatusConflict { |
| + t.Fatalf("cut from preparing = %d %s, want 409", code, out) |
| + } |
| + // Abort from preparing → 202 aborting (two-phase: the SOURCE box |
| + // deletes staging, then clears the field). |
| + if code, _, out := rawReq(t, http.MethodDelete, moveURL, token, ""); code != http.StatusAccepted || !strings.Contains(out, provision.MoveAborting) { |
| + t.Fatalf("abort = %d %s, want 202 aborting", code, out) |
| + } |
| + // A second abort while aborting → 409; a new request while aborting → |
| + // 409 (still in flight). |
| + if code, _, _ := rawReq(t, http.MethodDelete, moveURL, token, ""); code != http.StatusConflict { |
| + t.Fatalf("double abort = %d, want 409", code) |
| + } |
| + if code, _, _ := rawReq(t, http.MethodPost, moveURL, token, `{"region":"uk"}`); code != http.StatusConflict { |
| + t.Fatalf("request while aborting = %d, want 409", code) |
| + } |
| + // The box's half clears it; then a fresh request proceeds. |
| + if err := provision.SetRecordMove(context.Background(), fx.rel, "bab", "titogo", "acme.go.tito.io", provision.MoveAbortComplete()); err != nil { |
| + t.Fatal(err) |
| + } |
| + code, _, out = rawReq(t, http.MethodGet, u+"/acme.go.tito.io", token, "") |
| + if code != http.StatusOK || strings.Contains(out, `"move"`) { |
| + t.Fatalf("after abort-complete: %d %s, want no move key", code, out) |
| + } |
| + if code, _, _ := rawReq(t, http.MethodPost, moveURL, token, `{"region":"uk"}`); code != http.StatusAccepted { |
| + t.Fatal("re-request") |
| + } |
| + if err := provision.SetRecordMove(context.Background(), fx.rel, "bab", "titogo", "acme.go.tito.io", provision.MovePromoteReady(time.Now())); err != nil { |
| + t.Fatal(err) |
| + } |
| + if code, _, out := rawReq(t, http.MethodPost, moveURL+"/cut", token, ""); code != http.StatusAccepted { |
| + t.Fatalf("cut from ready = %d %s, want 202", code, out) |
| + } |
| + if code, _, out := rawReq(t, http.MethodDelete, moveURL, token, ""); code != http.StatusConflict { |
| + t.Fatalf("abort from cutting = %d %s, want 409 (completes forward)", code, out) |
| + } |
| + // The serial was bumped by the successful writes. |
| + if _, _, err := fx.rel.Get(context.Background(), "control/serial"); err != nil { |
| + t.Fatalf("control/serial not written: %v", err) |
| + } |
| + // 404 discipline: unknown host is the byte-identical not-found. |
| + if code, _, _ := rawReq(t, http.MethodPost, u+"/ghost.go.tito.io/move", token, `{"region":"uk"}`); code != http.StatusNotFound { |
| + t.Fatalf("move on unknown host = %d, want 404", code) |
| + } |
| +} |
| + |
| +// The class rails: only an exec-backed instance route moves. |
| +func TestAPIInstanceMoveClassRails(t *testing.T) { |
| + fx, token := instanceFixture(t, "code-inst-move-rails") |
| + m := titoManifest() |
| + m.Regions = nil |
| + m.Pool = "" |
| + enableInstances(t, fx, "titogo", m) |
| + u := instancesURL(fx, "bab", "titogo") |
| + if _, err := fx.rel.Put(context.Background(), "control/edges/uk-box.json", |
| + []byte(`{"v":1,"box":"uk-box","eip":"1.2.3.4","placement":"uk","hosts":{}}`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + mk := func(body string) string { |
| + t.Helper() |
| + code, _, out := rawReq(t, http.MethodPost, u, token, body) |
| + if code != http.StatusCreated { |
| + t.Fatalf("create %s = %d %s", body, code, out) |
| + } |
| + var rec provision.Record |
| + if err := json.Unmarshal([]byte(out), &rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + return rec.Host |
| + } |
| + for _, tc := range []struct { |
| + name, body, wantIn string |
| + }{ |
| + {"static", `{"host":"st.go.tito.io","kind":"static"}`, "kind static"}, |
| + {"addr-backed", `{"host":"ad.go.tito.io","addrs":["10.0.0.1:80"],"kind":"service"}`, "kind service"}, |
| + } { |
| + t.Run(tc.name, func(t *testing.T) { |
| + host := mk(tc.body) |
| + code, _, out := rawReq(t, http.MethodPost, u+"/"+host+"/move", token, `{"region":"uk"}`) |
| + if code != http.StatusBadRequest || !strings.Contains(out, tc.wantIn) { |
| + t.Fatalf("move %s = %d %s, want 400 mentioning %q", tc.name, code, out, tc.wantIn) |
| + } |
| + }) |
| + } |
| + // Fleet-placed: region carries fleet/<name>. |
| + host := mk(`{"host":"fl.go.tito.io"}`) |
| + if err := provision.SetRecordRegion(context.Background(), fx.rel, "bab", "titogo", host, "fleet/pi"); err != nil { |
| + t.Fatal(err) |
| + } |
| + code, _, out := rawReq(t, http.MethodPost, u+"/"+host+"/move", token, `{"region":"uk"}`) |
| + if code != http.StatusBadRequest || !strings.Contains(out, "fleet") { |
| + t.Fatalf("move fleet-placed = %d %s, want 400 mentioning fleet", code, out) |
| + } |
| + _ = fmt.Sprint() |
| + _ = provisionPooledMoveRefusal(t, fx, token, u) |
| +} |
| + |
| +// provisionPooledMoveRefusal pins the pool rail: a pooled record's data |
| +// lives on the pool's box, and v1 refuses the class at the door. |
| +func provisionPooledMoveRefusal(t *testing.T, fx *consoleFixture, token, u string) bool { |
| + t.Helper() |
| + rec := provision.Record{Schema: 1, Host: "pooled.go.tito.io", Slug: "pooled", App: "titogo", |
| + Account: "bab", Pool: "dedicated-x", Channel: "stable", CreatedAt: time.Now().UTC()} |
| + if err := provision.CreateRecord(context.Background(), fx.rel, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + code, _, out := rawReq(t, http.MethodPost, u+"/pooled.go.tito.io/move", token, `{"region":"uk"}`) |
| + if code != http.StatusBadRequest || !strings.Contains(out, "pool") { |
| + t.Fatalf("move pooled = %d %s, want 400 mentioning the pool rail", code, out) |
| + } |
| + return true |
| +} |
diff --git a/internal/console/authcli.go b/internal/console/authcli.go| index 14eab60b2..4bb4f6bc4 100644 |
| --- a/internal/console/authcli.go |
| +++ b/internal/console/authcli.go |
| @@ -583,6 +583,14 @@ func (s *Server) registerCLIRoutes() { |
| // keeps serving. Same {host} wildcard, same gate, same byte-identical |
| // 404 as its GET and DELETE siblings. |
| s.cliRoute("PATCH /api/cli/apps/{account}/{app}/instances/{host}", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceUpdate) |
| + // The residence-move doors (issue #331): requested by the tenant's |
| + // ROUTER with the app-held instance token (titogo's residence design — |
| + // never by an instance about itself... which the role system cannot yet |
| + // express; the Instance role is the same authority the create and |
| + // delete doors already grant this credential). |
| + s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/move", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceMoveRequest) |
| + s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/move/cut", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceMoveCut) |
| + s.cliRoute("DELETE /api/cli/apps/{account}/{app}/instances/{host}/move", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceMoveAbort) |
| s.cliRoute("DELETE /api/cli/apps/{account}/{app}/instances/{host}", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceDelete) |
| // The zero Set, deliberately: a credential may never mint a credential, |
| // including itself. This is the escalation cut (spec §3). It takes no |
diff --git a/internal/console/instancesplacement.go b/internal/console/instancesplacement.go| index 649325f8d..9bebccaa8 100644 |
| --- a/internal/console/instancesplacement.go |
| +++ b/internal/console/instancesplacement.go |
| @@ -33,12 +33,14 @@ type servedBy struct { |
| // capacity, and "where is this instance" is the question, not "who is its |
| // write home" (that is the home map's question, internal/edge). |
| type edgesFilePlacementView struct { |
| - V int `json:"v"` |
| - Box string `json:"box"` |
| - EIP string `json:"eip"` |
| - Region string `json:"region"` |
| - Placement string `json:"placement"` |
| - Hosts map[string]json.RawMessage `json:"hosts"` |
| + V int `json:"v"` |
| + Box string `json:"box"` |
| + EIP string `json:"eip"` |
| + Region string `json:"region"` |
| + Placement string `json:"placement"` |
| + DataBucket string `json:"data_bucket"` |
| + DataBucketRegion string `json:"data_bucket_region"` |
| + Hosts map[string]json.RawMessage `json:"hosts"` |
| } |
| // loadEdgesJoin folds every readable edges file into host → the boxes that |
diff --git a/internal/console/svcroutes_test.go b/internal/console/svcroutes_test.go| index c7fd77ac8..59cd0aeaf 100644 |
| --- a/internal/console/svcroutes_test.go |
| +++ b/internal/console/svcroutes_test.go |
| @@ -222,10 +222,21 @@ func TestNoRoleReachesAMintFleetOrPolicyDoor(t *testing.T) { |
| // the domains list answer that role has always reached, so this discloses |
| // nothing the list does not. The WRITE beside it is admin-only, and |
| // TestAPIFeaturesRoles pins that an instance credential never reaches it. |
| +// The three move doors (#331) are the fourth, fifth and sixth, together: |
| +// they did not exist before service principals, so listing them in `want` |
| +// would make the legacy map lie. The instance role reaches them |
| +// DELIBERATELY — the tenant's ROUTER drives a move with the app-held |
| +// CARLOS_INSTANCE_TOKEN (the residence design's stated caller), and a |
| +// credential that may already CREATE and DELETE instances (both in `want`) |
| +// is not gaining authority by relocating one. TestAPIInstanceMoveVerbs |
| +// pins what the doors themselves refuse. |
| var exemptFromLegacyPin = map[string]bool{ |
| "GET /api/cli/apps": true, |
| - "GET /api/cli/apps/{account}/{app}/domains/{host}": true, |
| - "GET /api/cli/apps/{account}/{app}/features": true, |
| + "GET /api/cli/apps/{account}/{app}/domains/{host}": true, |
| + "GET /api/cli/apps/{account}/{app}/features": true, |
| + "POST /api/cli/apps/{account}/{app}/instances/{host}/move": true, |
| + "POST /api/cli/apps/{account}/{app}/instances/{host}/move/cut": true, |
| + "DELETE /api/cli/apps/{account}/{app}/instances/{host}/move": true, |
| } |
| // TestInstanceRoleMatchesTodaysAppTokenReach is the compatibility pin of spec |
diff --git a/internal/host/agent.go b/internal/host/agent.go| index 13bb1580c..c6ead7c4f 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -307,6 +307,18 @@ func RunAgent(args []string) { |
| if logSink != nil { |
| svc.Logs = logSink |
| } |
| + // The residence-move machinery (issue #331), armed with the |
| + // activator: the staging-fallback restore needs a same-region |
| + // mirror client, and the move worker needs the service's quiesce |
| + // answer. Both are best-effort — a mirror client that cannot |
| + // build logs and leaves the fallback dark, exactly like every |
| + // other optional duty here. |
| + if m, err := lease.NewS3Mirror(context.Background(), cfg.Region, cfg.Region); err != nil { |
| + log.Printf("agent: staging-fallback mirror client: %v (restore fallback disarmed)", err) |
| + } else { |
| + eb.Mirror = m |
| + } |
| + armMovesPass(*cfg, dd, reg, svc) |
| svc.Start() |
| act = svc |
| log.Printf("agent: hibernation activator armed (bucket %s, idle %s)", cfg.Bucket, cfg.Idle) |
| @@ -1451,6 +1463,7 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu |
| if err := reconcileSerialized(ctx, store, reg, provisionTickConfig(os.Getenv)); err != nil { |
| log.Printf("agent: fast converge reconcile: %v", err) |
| } |
| + runMovesPass(ctx, store) |
| // Folded once per pass and handed in, never re-read inside the |
| // loop: this leg runs on every serial bump (a seconds cadence), |
| // and the fold is a LoadBoxes plus a List plus one Get per edges |
| @@ -1916,6 +1929,7 @@ func instancesLoop(reg *registry.Registry) { |
| if err := reconcileSerialized(ctx, store, reg, cfg); err != nil { |
| log.Printf("agent: instances pass: %v", err) |
| } |
| + runMovesPass(ctx, store) |
| deploymentRoutes := edgesfile.DeploymentRoutes(ctx, store, time.Now()) |
| if err := domains.ReconcileAliases(ctx, store, reg, time.Now, deploymentRoutes); err != nil { |
| log.Printf("agent: instances alias pass: %v", err) |
diff --git a/internal/host/edgesfile.go b/internal/host/edgesfile.go| index d57bbd1aa..dd4ec7685 100644 |
| --- a/internal/host/edgesfile.go |
| +++ b/internal/host/edgesfile.go |
| @@ -85,9 +85,17 @@ type edgesFile struct { |
| // console's create door folds every file's Placement into the armed set |
| // a requested instance region is validated against — dormant until at |
| // least one box publishes the field, so roll order does not matter. |
| - Placement string `json:"placement,omitempty"` |
| - Generation string `json:"generation"` |
| - Hosts map[string]edgeHost `json:"hosts"` |
| + Placement string `json:"placement,omitempty"` |
| + // DataBucket/DataBucketRegion are this box's hibernate/data bucket |
| + // coordinates (CARLOS_HIBERNATE_BUCKET/_REGION verbatim) — published so |
| + // a move SOURCE can aim its staging mirror at the target box's bucket |
| + // from the same file set everything else reads (instance-move spec). |
| + // Additive + omitempty like Placement; a box with no hibernate bucket |
| + // (activator disabled) publishes neither key. |
| + DataBucket string `json:"data_bucket,omitempty"` |
| + DataBucketRegion string `json:"data_bucket_region,omitempty"` |
| + Generation string `json:"generation"` |
| + Hosts map[string]edgeHost `json:"hosts"` |
| } |
| // edgesFilePass builds this box's edges file from the routes currently in |
| @@ -123,7 +131,7 @@ type edgesFile struct { |
| // once per alias per process (cleared when the alias publishes). |
| var withheldAliasSaid sync.Map |
| -func edgesFilePass(ctx context.Context, s release.Store, reg *registry.Registry, box, eip, region, placement string, last *[]byte) error { |
| +func edgesFilePass(ctx context.Context, s release.Store, reg *registry.Registry, box, eip, region, placement string, dataBucket, dataBucketRegion string, last *[]byte) error { |
| routes, err := reg.List() |
| if err != nil { |
| return fmt.Errorf("edges: list: %w", err) |
| @@ -257,7 +265,8 @@ func edgesFilePass(ctx context.Context, s release.Store, reg *registry.Registry, |
| } |
| } |
| - f := edgesFile{V: 1, Box: box, EIP: eip, Region: region, Placement: placement, Hosts: hosts} |
| + f := edgesFile{V: 1, Box: box, EIP: eip, Region: region, Placement: placement, |
| + DataBucket: dataBucket, DataBucketRegion: dataBucketRegion, Hosts: hosts} |
| gate, err := json.Marshal(f) |
| if err != nil { |
| return fmt.Errorf("edges: marshal: %w", err) |
| @@ -372,6 +381,8 @@ func edgesFileLoop(reg *registry.Registry) { |
| // reconcile pass scopes by — the two vocabularies travel in one file so |
| // a reader can join them without a second source. |
| placement := strings.TrimSpace(os.Getenv("CARLOS_REGION")) |
| + dataBucket := strings.TrimSpace(os.Getenv("CARLOS_HIBERNATE_BUCKET")) |
| + dataBucketRegion := strings.TrimSpace(os.Getenv("CARLOS_HIBERNATE_REGION")) |
| log.Printf("agent: edges file loop armed (5min + jitter, %s)", edgesFileKey(box)) |
| last := seedEdgesFileGate(ctx, store, box) |
| delay := time.Minute |
| @@ -379,7 +390,7 @@ func edgesFileLoop(reg *registry.Registry) { |
| time.Sleep(delay) |
| delay = 5*time.Minute + rand.N(time.Minute) |
| eip := readPublicIP() |
| - if err := edgesFilePass(ctx, store, reg, box, eip, region, placement, &last); err != nil { |
| + if err := edgesFilePass(ctx, store, reg, box, eip, region, placement, dataBucket, dataBucketRegion, &last); err != nil { |
| log.Printf("agent: edges file pass: %v", err) |
| } |
| } |
diff --git a/internal/host/edgesfile_test.go b/internal/host/edgesfile_test.go| index 2c219db67..db859c8e8 100644 |
| --- a/internal/host/edgesfile_test.go |
| +++ b/internal/host/edgesfile_test.go |
| @@ -55,7 +55,7 @@ func TestEdgesFilePublishesHomedHosts(t *testing.T) { |
| s, reg := release.NewDirStore(t.TempDir()), edgesTestRegistry(t) |
| seedRoute(t, reg, "app.example.com") |
| var lastBody []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", &lastBody); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", "", "", &lastBody); err != nil { |
| t.Fatal(err) |
| } |
| got, _, err := s.Get(context.Background(), "control/edges/annie.json") |
| @@ -86,7 +86,7 @@ func TestEdgesFilePublishesWithEmptyEIP(t *testing.T) { |
| s, reg := release.NewDirStore(t.TempDir()), edgesTestRegistry(t) |
| seedRoute(t, reg, "app.example.com") |
| var lastBody []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "", "", "", &lastBody); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "", "", "", "", "", &lastBody); err != nil { |
| t.Fatal(err) |
| } |
| got, _, err := s.Get(context.Background(), "control/edges/annie.json") |
| @@ -112,7 +112,7 @@ func TestEdgesFileSkipsWriteWhenUnchanged(t *testing.T) { |
| seedRoute(t, reg, "app.example.com") |
| var lastBody []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", &lastBody); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", "", "", &lastBody); err != nil { |
| t.Fatal(err) |
| } |
| if s.puts != 1 { |
| @@ -122,7 +122,7 @@ func TestEdgesFileSkipsWriteWhenUnchanged(t *testing.T) { |
| // Second pass, identical registry state: a fresh generation timestamp |
| // would differ from the first pass's, but the change-gate excludes |
| // Generation, so this must not write. |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", &lastBody); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", "", "", &lastBody); err != nil { |
| t.Fatal(err) |
| } |
| if s.puts != 1 { |
| @@ -131,7 +131,7 @@ func TestEdgesFileSkipsWriteWhenUnchanged(t *testing.T) { |
| // A registry change makes the next pass write again. |
| seedRoute(t, reg, "second.example.com") |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", &lastBody); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", "", "", &lastBody); err != nil { |
| t.Fatal(err) |
| } |
| if s.puts != 2 { |
| @@ -151,12 +151,12 @@ func TestEdgesFileHoldsLastKnownGoodEIPOnTransientBlank(t *testing.T) { |
| seedRoute(t, reg, "app.example.com") |
| var last []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", &last); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", "", "", &last); err != nil { |
| t.Fatal(err) |
| } |
| // Simulate a transient IMDS blip: this tick's probe returns "". |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "", "", "", &last); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "", "", "", "", "", &last); err != nil { |
| t.Fatal(err) |
| } |
| @@ -186,7 +186,7 @@ func TestEdgesFileExcludesNonReadyRows(t *testing.T) { |
| } |
| var last []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", &last); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", "", "", &last); err != nil { |
| t.Fatal(err) |
| } |
| got, _, err := s.Get(context.Background(), "control/edges/annie.json") |
| @@ -219,7 +219,7 @@ func TestEdgesFileWildcardRowPublishesVerbatim(t *testing.T) { |
| } |
| var last []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", &last); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", "", "", &last); err != nil { |
| t.Fatal(err) |
| } |
| got, _, err := s.Get(context.Background(), "control/edges/annie.json") |
| @@ -250,7 +250,7 @@ func TestEdgesFileRawWireBodyMatchesSchema(t *testing.T) { |
| seedRoute(t, reg, "app.example.com") |
| var last []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "quiet-harbour", "203.0.113.10", "", "", &last); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "quiet-harbour", "203.0.113.10", "", "", "", "", &last); err != nil { |
| t.Fatal(err) |
| } |
| got, _, err := s.Get(context.Background(), "control/edges/quiet-harbour.json") |
| @@ -302,7 +302,7 @@ func TestEdgesFileRawWireBodyRegionField(t *testing.T) { |
| t.Run("unset region is omitted entirely", func(t *testing.T) { |
| var last []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "quiet-harbour", "203.0.113.10", "", "", &last); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "quiet-harbour", "203.0.113.10", "", "", "", "", &last); err != nil { |
| t.Fatal(err) |
| } |
| got, _, err := s.Get(context.Background(), "control/edges/quiet-harbour.json") |
| @@ -321,7 +321,7 @@ func TestEdgesFileRawWireBodyRegionField(t *testing.T) { |
| s2, reg2 := release.NewDirStore(t.TempDir()), edgesTestRegistry(t) |
| seedRoute(t, reg2, "app.example.com") |
| var last []byte |
| - if err := edgesFilePass(context.Background(), s2, reg2, "quiet-harbour", "203.0.113.10", "eu-west-1", "", &last); err != nil { |
| + if err := edgesFilePass(context.Background(), s2, reg2, "quiet-harbour", "203.0.113.10", "eu-west-1", "", "", "", &last); err != nil { |
| t.Fatal(err) |
| } |
| got, _, err := s2.Get(context.Background(), "control/edges/quiet-harbour.json") |
| @@ -423,7 +423,7 @@ func TestEdgesFileSeedGateSkipsWriteOnUnchangedBucketState(t *testing.T) { |
| // A prior process (or the prior pass of this one) already published. |
| var priming []byte |
| - if err := edgesFilePass(context.Background(), real, reg, "annie", "1.2.3.4", "", "", &priming); err != nil { |
| + if err := edgesFilePass(context.Background(), real, reg, "annie", "1.2.3.4", "", "", "", "", &priming); err != nil { |
| t.Fatal(err) |
| } |
| @@ -435,7 +435,7 @@ func TestEdgesFileSeedGateSkipsWriteOnUnchangedBucketState(t *testing.T) { |
| } |
| counting := &putCountingStore{Store: real} |
| - if err := edgesFilePass(context.Background(), counting, reg, "annie", "1.2.3.4", "", "", &last); err != nil { |
| + if err := edgesFilePass(context.Background(), counting, reg, "annie", "1.2.3.4", "", "", "", "", &last); err != nil { |
| t.Fatal(err) |
| } |
| if counting.puts != 0 { |
| @@ -454,7 +454,7 @@ func TestEdgesFileSeedHoldsEIPThroughRestartBlip(t *testing.T) { |
| seedRoute(t, reg, "app.example.com") |
| var priming []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", &priming); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", "", "", &priming); err != nil { |
| t.Fatal(err) |
| } |
| @@ -466,7 +466,7 @@ func TestEdgesFileSeedHoldsEIPThroughRestartBlip(t *testing.T) { |
| // This tick both hits a transient IMDS blip AND sees a real registry |
| // change, so the pass must write. |
| seedRoute(t, reg, "second.example.com") |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "", "", "", &last); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "", "", "", "", "", &last); err != nil { |
| t.Fatal(err) |
| } |
| @@ -495,13 +495,13 @@ func TestEdgesFileUnseededLastPublishesBlankEIPOnFirstTickBlip(t *testing.T) { |
| seedRoute(t, reg, "app.example.com") |
| var priming []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", &priming); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", "", "", &priming); err != nil { |
| t.Fatal(err) |
| } |
| // The pre-fix shape: a fresh process's *last starts nil (no seed read). |
| var unseeded []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "", "", "", &unseeded); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "", "", "", "", "", &unseeded); err != nil { |
| t.Fatal(err) |
| } |
| @@ -545,7 +545,7 @@ func TestEdgesFileSeedPreRegionBucketWritesOnceWhenRegionArrives(t *testing.T) { |
| // This build now has CARLOS_REGION set: the first pass must write, once. |
| counting := &putCountingStore{Store: real} |
| - if err := edgesFilePass(context.Background(), counting, reg, "annie", "1.2.3.4", "eu-west-1", "", &last); err != nil { |
| + if err := edgesFilePass(context.Background(), counting, reg, "annie", "1.2.3.4", "eu-west-1", "", "", "", &last); err != nil { |
| t.Fatal(err) |
| } |
| if counting.puts != 1 { |
| @@ -565,7 +565,7 @@ func TestEdgesFileSeedPreRegionBucketWritesOnceWhenRegionArrives(t *testing.T) { |
| // A second, identical-region pass must not write again — the gate now |
| // includes region on both sides. |
| - if err := edgesFilePass(context.Background(), counting, reg, "annie", "1.2.3.4", "eu-west-1", "", &last); err != nil { |
| + if err := edgesFilePass(context.Background(), counting, reg, "annie", "1.2.3.4", "eu-west-1", "", "", "", &last); err != nil { |
| t.Fatal(err) |
| } |
| if counting.puts != 1 { |
| @@ -588,7 +588,7 @@ func TestEdgesFileMarksPlatformRows(t *testing.T) { |
| } |
| } |
| var lastBody []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", &lastBody); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", "", "", &lastBody); err != nil { |
| t.Fatal(err) |
| } |
| body, _, err := s.Get(context.Background(), "control/edges/annie.json") |
| @@ -654,7 +654,7 @@ func TestEdgesFilePublishesPointingAliasesOfServedPrimaries(t *testing.T) { |
| publish := func() map[string]hostView { |
| t.Helper() |
| var lastBody []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", &lastBody); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", "", "", &lastBody); err != nil { |
| t.Fatal(err) |
| } |
| body, _, err := s.Get(context.Background(), "control/edges/annie.json") |
| @@ -734,7 +734,7 @@ func TestEdgesFilePublishesDelegatedWildcardParents(t *testing.T) { |
| } |
| } |
| var lastBody []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", &lastBody); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "annie", "1.2.3.4", "", "", "", "", &lastBody); err != nil { |
| t.Fatal(err) |
| } |
| body, _, err := s.Get(context.Background(), "control/edges/annie.json") |
| @@ -797,7 +797,7 @@ func TestEdgesFileBoxNameJoinsTheStatusSnapshotLabel(t *testing.T) { |
| s, reg := release.NewDirStore(t.TempDir()), edgesTestRegistry(t) |
| seedRoute(t, reg, "app.example.com") |
| var last []byte |
| - if err := edgesFilePass(ctx, s, reg, box, "203.0.113.7", "eu-west-1", "", &last); err != nil { |
| + if err := edgesFilePass(ctx, s, reg, box, "203.0.113.7", "eu-west-1", "", "", "", &last); err != nil { |
| t.Fatal(err) |
| } |
| if err := status.WriteBox(ctx, s, snapshot); err != nil { |
| @@ -817,7 +817,7 @@ func TestEdgesFileRawWireBodyPlacementField(t *testing.T) { |
| s, reg := release.NewDirStore(t.TempDir()), edgesTestRegistry(t) |
| seedRoute(t, reg, "app.example.com") |
| var last []byte |
| - if err := edgesFilePass(context.Background(), s, reg, "quiet-harbour", "203.0.113.10", "eu-west-2", "uk", &last); err != nil { |
| + if err := edgesFilePass(context.Background(), s, reg, "quiet-harbour", "203.0.113.10", "eu-west-2", "uk", "", "", &last); err != nil { |
| t.Fatal(err) |
| } |
| got, _, err := s.Get(context.Background(), "control/edges/quiet-harbour.json") |
| @@ -841,7 +841,7 @@ func TestEdgesFileRawWireBodyPlacementField(t *testing.T) { |
| s2, reg2 := release.NewDirStore(t.TempDir()), edgesTestRegistry(t) |
| seedRoute(t, reg2, "app.example.com") |
| var last2 []byte |
| - if err := edgesFilePass(context.Background(), s2, reg2, "quiet-harbour", "203.0.113.10", "", "", &last2); err != nil { |
| + if err := edgesFilePass(context.Background(), s2, reg2, "quiet-harbour", "203.0.113.10", "", "", "", "", &last2); err != nil { |
| t.Fatal(err) |
| } |
| got2, _, err := s2.Get(context.Background(), "control/edges/quiet-harbour.json") |
| @@ -852,3 +852,42 @@ func TestEdgesFileRawWireBodyPlacementField(t *testing.T) { |
| t.Fatalf("raw wire body with no CARLOS_REGION carries a placement key: %s", got2) |
| } |
| } |
| + |
| +// Data-bucket coordinates (instance-move spec): published verbatim from the |
| +// hibernate env pair, absent when unset — the same additive rule as |
| +// placement. |
| +func TestEdgesFileRawWireBodyDataBucketFields(t *testing.T) { |
| + s, reg := release.NewDirStore(t.TempDir()), edgesTestRegistry(t) |
| + seedRoute(t, reg, "app.example.com") |
| + var last []byte |
| + if err := edgesFilePass(context.Background(), s, reg, "quiet-harbour", "203.0.113.10", "eu-west-2", "uk", "carlos-tito-uk-1", "eu-west-2", &last); err != nil { |
| + t.Fatal(err) |
| + } |
| + got, _, err := s.Get(context.Background(), "control/edges/quiet-harbour.json") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var probe struct { |
| + DataBucket string `json:"data_bucket"` |
| + DataBucketRegion string `json:"data_bucket_region"` |
| + } |
| + if err := json.Unmarshal(got, &probe); err != nil { |
| + t.Fatal(err) |
| + } |
| + if probe.DataBucket != "carlos-tito-uk-1" || probe.DataBucketRegion != "eu-west-2" { |
| + t.Fatalf("data bucket fields = %+v", probe) |
| + } |
| + s2, reg2 := release.NewDirStore(t.TempDir()), edgesTestRegistry(t) |
| + seedRoute(t, reg2, "app.example.com") |
| + var last2 []byte |
| + if err := edgesFilePass(context.Background(), s2, reg2, "quiet-harbour", "203.0.113.10", "", "", "", "", &last2); err != nil { |
| + t.Fatal(err) |
| + } |
| + got2, _, err := s2.Get(context.Background(), "control/edges/quiet-harbour.json") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if strings.Contains(string(got2), "data_bucket") { |
| + t.Fatalf("unset hibernate env still published a data_bucket key: %s", got2) |
| + } |
| +} |
diff --git a/internal/host/movespass.go b/internal/host/movespass.go| new file mode 100644 |
| index 000000000..1772374c5 |
| --- /dev/null |
| +++ b/internal/host/movespass.go |
| @@ -0,0 +1,300 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "fmt" |
| + "log" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "sync" |
| + "sync/atomic" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// The residence-move worker (issue #331, instance-move spec): everything a |
| +// move does to DATA happens here, on the SOURCE box — the box whose |
| +// CARLOS_REGION matches the record's region for as long as the move is in |
| +// flight. The console only writes requests; the target box only restores. |
| +// |
| +// One pass = one delta round per in-flight move. The caller runs it on the |
| +// same trigger as the reconcile pass (serial bumps land in seconds via |
| +// carlos-adopt.path), and prepare's convergence is the round-over-round |
| +// shrinking of the delta, not a resident goroutine — restart-safe because |
| +// every step derives from the record and the buckets. |
| +// |
| +// The cut ordering is the never-two-writers argument, mechanically: |
| +// |
| +// route Status "moving" (edge stops serving; RetireExcept stops the child) |
| +// → child quiesced (Quiesced(host) — retries next pass until true) |
| +// → final mirror round (own replica is byte-final; staging matches it) |
| +// → local purge (the warm-cache trap: a stale local file must |
| +// never resurrect a moved-away history) |
| +// → CAS flip+clear (Region=To, Move=nil, ONE write — the last act) |
| +// |
| +// Every step is idempotent; a box that dies mid-cut resumes at the top and |
| +// falls through the already-done steps. |
| + |
| +// armedMoveCfg is RunAgent's one write; runMovesPass' reads are its only |
| +// other touch. An atomic pointer rather than plain assignment because the |
| +// arm happens on RunAgent's goroutine and the passes run on the converge |
| +// and instances loops. |
| +var armedMoveCfg atomic.Pointer[moveCfg] |
| + |
| +// armMovesPass wires the worker with what only RunAgent has in hand: the |
| +// activator's hibernate coordinates (the own-bucket side of every mirror), |
| +// the data dir the cut purges, the registry the cut quiesces through, and |
| +// the service that answers whether a child is down. Called once, before |
| +// svc.Start. |
| +func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry, svc *activator.Service) { |
| + cfg := moveCfg{ |
| + Region: strings.TrimSpace(os.Getenv("CARLOS_REGION")), |
| + OwnBucket: hib.Bucket, |
| + OwnBucketRegion: hib.Region, |
| + OwnPrefix: hib.Prefix, |
| + DataDir: dataDir, |
| + Mirror: func(ctx context.Context, dstRegion string) (lease.MirrorStore, error) { |
| + return lease.NewS3Mirror(ctx, hib.Region, dstRegion) |
| + }, |
| + Quiesced: svc.Quiesced, |
| + SetStatus: reg.SetStatus, |
| + OwnsRoute: func(host string) bool { |
| + rt, err := reg.Get(host) |
| + return err == nil && rt != nil && rt.Provisioned |
| + }, |
| + Checkpoint: func(ctx context.Context, dbPath string) error { |
| + host := strings.TrimSuffix(filepath.Base(dbPath), ".db") |
| + rep, err := lease.ReplicateWithLease(ctx, dbPath, hib.Bucket, activator.HibernatePrefix(hib.Prefix, host), hib.Region) |
| + if err != nil { |
| + return err |
| + } |
| + defer rep.Close(ctx) |
| + return rep.SyncAndWait(ctx) |
| + }, |
| + Now: time.Now, |
| + } |
| + armedMoveCfg.Store(&cfg) |
| +} |
| + |
| +// runMovesPass is the loops' call: inert until armed (no activator = no |
| +// exec instances = nothing this box could ever move). |
| +func runMovesPass(ctx context.Context, s release.Store) { |
| + cfg := armedMoveCfg.Load() |
| + if cfg == nil || s == nil { |
| + return |
| + } |
| + if err := movesPass(ctx, s, *cfg); err != nil { |
| + log.Printf("agent: moves pass: %v", err) |
| + } |
| +} |
| + |
| +// moveCfg carries what the worker needs, injectable for tests. |
| +type moveCfg struct { |
| + Region string // this box's placement code, "" included (the home) |
| + OwnBucket string // CARLOS_HIBERNATE_BUCKET |
| + OwnBucketRegion string // CARLOS_HIBERNATE_REGION |
| + OwnPrefix string // CARLOS_HIBERNATE_PREFIX ("instances" in production) |
| + DataDir string // /data — where the local DB lives |
| + Mirror func(ctx context.Context, dstRegion string) (lease.MirrorStore, error) |
| + Quiesced func(host string) bool |
| + SetStatus func(host, status string) error |
| + // OwnsRoute answers whether THIS box's registry serves the host |
| + // (review round 1): region equality alone would let a standby or |
| + // second same-region box drive a cut whose child runs on its sibling — |
| + // each box has its OWN registry, so the sibling's Quiesced is |
| + // trivially, wrongly true. The box that owns the route owns the move. |
| + OwnsRoute func(host string) bool |
| + // Checkpoint re-replicates the LOCAL DB to the own bucket and waits |
| + // (review round 1): the activator's stop logs — and swallows — a failed |
| + // final checkpoint, because pre-move the local warm cache made that |
| + // safe. The cut deletes the warm cache, so it must first prove the own |
| + // replica really carries the tail. Nil skips (tests without S3). |
| + Checkpoint func(ctx context.Context, dbPath string) error |
| + Now func() time.Time |
| +} |
| + |
| +// moveTarget is one box's coordinates as its edges file publishes them. |
| +type moveTarget struct { |
| + Placement string `json:"placement"` |
| + DataBucket string `json:"data_bucket"` |
| + DataBucketRegion string `json:"data_bucket_region"` |
| + EIP string `json:"eip"` |
| + V int `json:"v"` |
| + Box string `json:"box"` |
| +} |
| + |
| +// movesPass runs one round over every in-flight move this box owns. |
| +func movesPass(ctx context.Context, s release.Store, cfg moveCfg) error { |
| + recs, err := provision.ListAllRecords(ctx, s) |
| + if err != nil { |
| + return fmt.Errorf("moves: list records: %w", err) |
| + } |
| + var targets map[string]moveTarget // lazy: most passes carry no move |
| + for _, rec := range recs { |
| + if rec.Move == nil || rec.Region != cfg.Region { |
| + continue |
| + } |
| + // The rails the console's moveGate also holds (and this side must |
| + // re-hold — records have other writers): only an activator-managed |
| + // exec instance's state is this worker's to ship. A pooled record's |
| + // data lives on the pool's box, not this one (v1 refuses the whole |
| + // class), and a non-hibernating instance runs OUTSIDE the activator |
| + // — Quiesced would answer true over a live, writing process. |
| + if rec.Pool != "" || !rec.HibernateOrDefault() || (rec.Kind != "" && rec.Kind != "instance") { |
| + continue |
| + } |
| + if cfg.OwnsRoute != nil && !cfg.OwnsRoute(rec.Host) { |
| + continue // another box's child; its owner drives the move |
| + } |
| + if targets == nil { |
| + if targets, err = moveTargets(ctx, s); err != nil { |
| + log.Printf("moves: %v — every in-flight move waits for the next pass", err) |
| + return nil |
| + } |
| + } |
| + if err := runMoveRound(ctx, s, cfg, rec, targets); err != nil { |
| + // One move's trouble must not starve its neighbours — and a |
| + // PERSISTENT one (an unapplied staging grant's AccessDenied, |
| + // review round 1) must not flood the journal at the fast |
| + // converge loop's cadence: once a minute per host is plenty. |
| + logMoveTrouble(rec.Host, err) |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +// moveTargets folds the fleet's edges files into placement → coordinates. |
| +// A duplicate placement code keeps the lexically-first box (deterministic |
| +// everywhere, same rule as foldHomes) — and a target with no data bucket is |
| +// simply absent, which runMoveRound reports per-move. |
| +func moveTargets(ctx context.Context, s release.Store) (map[string]moveTarget, error) { |
| + bodies, err := loadEdgesFileBodies(ctx, s) |
| + if err != nil { |
| + return nil, err |
| + } |
| + out := map[string]moveTarget{} |
| + for _, body := range bodies { |
| + var t moveTarget |
| + if err := json.Unmarshal(body, &t); err != nil || t.V < 1 || t.Placement == "" || t.DataBucket == "" { |
| + continue |
| + } |
| + if prev, dup := out[t.Placement]; dup && prev.Box <= t.Box { |
| + continue |
| + } |
| + out[t.Placement] = t |
| + } |
| + return out, nil |
| +} |
| + |
| +func runMoveRound(ctx context.Context, s release.Store, cfg moveCfg, rec provision.Record, targets map[string]moveTarget) error { |
| + tgt, ok := targets[rec.Move.To] |
| + if !ok { |
| + return fmt.Errorf("target region %s publishes no data bucket yet — waiting", rec.Move.To) |
| + } |
| + m, err := cfg.Mirror(ctx, tgt.DataBucketRegion) |
| + if err != nil { |
| + return fmt.Errorf("mirror client for %s: %w", tgt.DataBucketRegion, err) |
| + } |
| + srcPrefix := activator.HibernatePrefix(cfg.OwnPrefix, rec.Host) |
| + dstPrefix := activator.MovesPrefix(rec.Host) |
| + |
| + switch rec.Move.State { |
| + case provision.MovePreparing, provision.MoveReady: |
| + changed, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix) |
| + if err != nil { |
| + return err |
| + } |
| + if rec.Move.State == provision.MovePreparing && changed == 0 { |
| + err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MovePromoteReady(cfg.Now())) |
| + if err != nil && !isMoveConflict(err) { |
| + return err |
| + } |
| + log.Printf("moves: %s: staged replica caught up in %s/%s — ready to cut", rec.Host, tgt.DataBucket, dstPrefix) |
| + } |
| + return nil |
| + |
| + case provision.MoveAborting: |
| + // Abort's box half: delete the staged copy, THEN clear the move — |
| + // a lingering staging would win a later landing on the target |
| + // (staging-wins, activator.promoteStagedReplica). |
| + staged, err := m.List(ctx, tgt.DataBucket, dstPrefix) |
| + if err != nil { |
| + return err |
| + } |
| + for k := range staged { |
| + if err := m.Delete(ctx, tgt.DataBucket, k); err != nil { |
| + return err |
| + } |
| + } |
| + err = provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveAbortComplete()) |
| + if err != nil && !isMoveConflict(err) { |
| + return err |
| + } |
| + log.Printf("moves: %s: abort complete — staged copy deleted, instance stays in %q", rec.Host, rec.Region) |
| + return nil |
| + |
| + case provision.MoveCutting: |
| + // Step 1+2: quiesce. SetStatus is idempotent; a route already |
| + // "moving" (a resumed cut) just re-writes the word. |
| + if err := cfg.SetStatus(rec.Host, "moving"); err != nil { |
| + return fmt.Errorf("set route status: %w", err) |
| + } |
| + if !cfg.Quiesced(rec.Host) { |
| + return nil // the retire is in flight; next pass re-enters here |
| + } |
| + // Step 3: prove the own replica carries the tail. The activator's |
| + // stop SWALLOWS a failed final checkpoint (safe pre-move: the warm |
| + // cache stood behind it; fatal here: step 5 deletes that cache). |
| + // Re-checkpointing an already-synced DB is cheap and idempotent; a |
| + // held lease (the child not fully released) or an S3 blip returns |
| + // an error and the pass retries. |
| + dbPath := filepath.Join(cfg.DataDir, rec.Host+".db") |
| + if cfg.Checkpoint != nil { |
| + if _, statErr := os.Stat(dbPath); statErr == nil { |
| + if err := cfg.Checkpoint(ctx, dbPath); err != nil { |
| + return fmt.Errorf("final checkpoint: %w", err) |
| + } |
| + } |
| + } |
| + // Step 4: the byte-final mirror round. |
| + if _, err := lease.MirrorPrefix(ctx, m, cfg.OwnBucket, srcPrefix, tgt.DataBucket, dstPrefix); err != nil { |
| + return err |
| + } |
| + // Step 5: the warm-cache purge. |
| + activator.PurgeLocalDBArtifacts(dbPath, true) |
| + // Step 6: the last act — Region flips, Move clears, one write. |
| + err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveComplete()) |
| + if err != nil && !isMoveConflict(err) { |
| + return err |
| + } |
| + log.Printf("moves: %s: cut complete — now homed in %s (route retires on the next reconcile pass)", rec.Host, rec.Move.To) |
| + return nil |
| + } |
| + return nil |
| +} |
| + |
| +// logMoveTrouble rate-limits per-host move errors to one line a minute. |
| +var moveTroubleLast sync.Map // host → time.Time |
| + |
| +func logMoveTrouble(host string, err error) { |
| + now := time.Now() |
| + if last, ok := moveTroubleLast.Load(host); ok && now.Sub(last.(time.Time)) < time.Minute { |
| + return |
| + } |
| + moveTroubleLast.Store(host, now) |
| + log.Printf("moves: %s: %v", host, err) |
| +} |
| + |
| +func isMoveConflict(err error) bool { |
| + // A lost CAS whose re-read finds the transition already made surfaces |
| + // as ErrMoveConflict: someone (this box, last life) already advanced |
| + // it. That is completion, not failure. |
| + return err != nil && strings.Contains(err.Error(), provision.ErrMoveConflict.Error()) |
| +} |
diff --git a/internal/host/movespass_test.go b/internal/host/movespass_test.go| new file mode 100644 |
| index 000000000..3772d3c9f |
| --- /dev/null |
| +++ b/internal/host/movespass_test.go |
| @@ -0,0 +1,294 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// fakeMoveMirror: bucket→key→etag plus an op log for ordering assertions. |
| +type fakeMoveMirror struct { |
| + buckets map[string]map[string]string |
| + ops []string |
| +} |
| + |
| +func newFakeMoveMirror() *fakeMoveMirror { |
| + return &fakeMoveMirror{buckets: map[string]map[string]string{}} |
| +} |
| +func (f *fakeMoveMirror) put(bucket, key, etag string) { |
| + if f.buckets[bucket] == nil { |
| + f.buckets[bucket] = map[string]string{} |
| + } |
| + f.buckets[bucket][key] = etag |
| +} |
| +func (f *fakeMoveMirror) List(_ context.Context, bucket, prefix string) (map[string]string, error) { |
| + out := map[string]string{} |
| + for k, v := range f.buckets[bucket] { |
| + if strings.HasPrefix(k, prefix) { |
| + out[k] = v |
| + } |
| + } |
| + return out, nil |
| +} |
| +func (f *fakeMoveMirror) Copy(_ context.Context, srcBucket, srcKey, dstBucket, dstKey string) error { |
| + f.ops = append(f.ops, "copy "+dstBucket+"/"+dstKey) |
| + f.put(dstBucket, dstKey, f.buckets[srcBucket][srcKey]) |
| + return nil |
| +} |
| +func (f *fakeMoveMirror) Delete(_ context.Context, bucket, key string) error { |
| + f.ops = append(f.ops, "delete "+bucket+"/"+key) |
| + delete(f.buckets[bucket], key) |
| + return nil |
| +} |
| + |
| +type moveHarness struct { |
| + s release.Store |
| + f *fakeMoveMirror |
| + cfg moveCfg |
| + statuses []string |
| + quiet bool |
| + owns bool |
| +} |
| + |
| +func newMoveHarness(t *testing.T) *moveHarness { |
| + t.Helper() |
| + h := &moveHarness{s: release.NewDirStore(t.TempDir()), f: newFakeMoveMirror(), quiet: true, owns: true} |
| + h.cfg = moveCfg{ |
| + Region: "ie", |
| + OwnBucket: "data-ie", |
| + OwnBucketRegion: "eu-west-1", |
| + OwnPrefix: "instances", |
| + DataDir: t.TempDir(), |
| + Mirror: func(_ context.Context, dstRegion string) (lease.MirrorStore, error) { |
| + return h.f, nil |
| + }, |
| + Quiesced: func(string) bool { return h.quiet }, |
| + SetStatus: func(host, status string) error { h.statuses = append(h.statuses, host+"="+status); return nil }, |
| + OwnsRoute: func(host string) bool { return h.owns }, |
| + Checkpoint: func(_ context.Context, dbPath string) error { |
| + h.f.ops = append(h.f.ops, "checkpoint "+filepath.Base(dbPath)) |
| + return nil |
| + }, |
| + Now: time.Now, |
| + } |
| + // The target's edges file: uk publishes a data bucket. |
| + if _, err := h.s.Put(context.Background(), "control/edges/uk-box.json", |
| + []byte(`{"v":1,"box":"uk-box","eip":"1.2.3.4","placement":"uk","data_bucket":"data-uk","data_bucket_region":"eu-west-2","hosts":{}}`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + return h |
| +} |
| + |
| +func (h *moveHarness) record(t *testing.T, host string, mut ...func(*provision.Record) error) provision.Record { |
| + t.Helper() |
| + rec := provision.Record{Schema: 1, Host: host, Slug: "s", App: "app", Account: "bab", |
| + Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()} |
| + if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, m := range mut { |
| + if err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, m); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + got, err := provision.GetRecord(context.Background(), h.s, "bab", "app", host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return *got |
| +} |
| + |
| +func (h *moveHarness) read(t *testing.T, host string) provision.Record { |
| + t.Helper() |
| + got, err := provision.GetRecord(context.Background(), h.s, "bab", "app", host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return *got |
| +} |
| + |
| +func TestMovesPassPrepareAndPromote(t *testing.T) { |
| + h := newMoveHarness(t) |
| + host := "a.go.example.com" |
| + h.record(t, host, provision.MoveRequest("uk", time.Now())) |
| + h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1") |
| + h.f.put("data-ie", "instances/"+host+"/generations/g/wal/000", "e2") |
| + |
| + // Round 1: copies happen; the SAME pass does not promote (the round |
| + // that copied cannot also claim caught-up — a write may have landed |
| + // mid-copy; only a round that finds NOTHING to do promotes). |
| + if err := movesPass(context.Background(), h.s, h.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := h.read(t, host); got.Move.State != provision.MovePreparing { |
| + t.Fatalf("after the copying round: %s, want still preparing", got.Move.State) |
| + } |
| + staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/") |
| + if len(staged) != 2 { |
| + t.Fatalf("staging = %v", staged) |
| + } |
| + // Round 2: nothing to do → ready. |
| + if err := movesPass(context.Background(), h.s, h.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + got := h.read(t, host) |
| + if got.Move.State != provision.MoveReady || got.Move.ReadyAt == nil { |
| + t.Fatalf("after the quiet round: %+v, want ready", got.Move) |
| + } |
| + // Ready keeps mirroring: a late write flows on the next round. |
| + h.f.put("data-ie", "instances/"+host+"/generations/g/wal/001", "e3") |
| + if err := movesPass(context.Background(), h.s, h.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + staged, _ = h.f.List(context.Background(), "data-uk", "moves/"+host+"/") |
| + if len(staged) != 3 || h.read(t, host).Move.State != provision.MoveReady { |
| + t.Fatalf("late write did not flow: %v", staged) |
| + } |
| +} |
| + |
| +func TestMovesPassCutSequence(t *testing.T) { |
| + h := newMoveHarness(t) |
| + host := "a.go.example.com" |
| + h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting()) |
| + h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1") |
| + // A local warm cache that must be purged. |
| + db := filepath.Join(h.cfg.DataDir, host+".db") |
| + if err := writeFileForTest(db, "stale"); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + // Not quiesced yet: the pass sets the status and stops there. |
| + h.quiet = false |
| + if err := movesPass(context.Background(), h.s, h.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(h.statuses) != 1 || h.statuses[0] != host+"=moving" { |
| + t.Fatalf("statuses = %v", h.statuses) |
| + } |
| + if got := h.read(t, host); got.Region != "ie" || got.Move.State != provision.MoveCutting { |
| + t.Fatalf("cut advanced past an un-quiesced child: %+v", got) |
| + } |
| + if len(h.f.ops) != 0 { |
| + t.Fatalf("mirror ran before quiesce: %v", h.f.ops) |
| + } |
| + |
| + // Quiesced: the whole tail runs, in order — the final own-bucket |
| + // checkpoint strictly BEFORE the mirror round (the mirror of a stale |
| + // replica is faithful and useless). |
| + h.quiet = true |
| + if err := movesPass(context.Background(), h.s, h.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + ckIdx, cpIdx := -1, -1 |
| + for i, op := range h.f.ops { |
| + if strings.HasPrefix(op, "checkpoint ") && ckIdx == -1 { |
| + ckIdx = i |
| + } |
| + if strings.HasPrefix(op, "copy ") && cpIdx == -1 { |
| + cpIdx = i |
| + } |
| + } |
| + if ckIdx == -1 || cpIdx == -1 || ckIdx > cpIdx { |
| + t.Fatalf("cut op order = %v, want checkpoint before the mirror copy", h.f.ops) |
| + } |
| + got := h.read(t, host) |
| + if got.Region != "uk" || got.Move != nil { |
| + t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move) |
| + } |
| + if fileExistsForTest(db) { |
| + t.Fatal("local DB survived the cut — the warm-cache trap") |
| + } |
| + staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/") |
| + if len(staged) != 1 { |
| + t.Fatalf("staging after final round = %v", staged) |
| + } |
| + // The pass is now inert for this record (region uk ≠ cfg ie). |
| + before := len(h.f.ops) |
| + if err := movesPass(context.Background(), h.s, h.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(h.f.ops) != before { |
| + t.Fatal("a completed move kept mirroring") |
| + } |
| +} |
| + |
| +func TestMovesPassSkipsWithoutTargetCoords(t *testing.T) { |
| + h := newMoveHarness(t) |
| + host := "a.go.example.com" |
| + h.record(t, host, provision.MoveRequest("se", time.Now())) // se publishes nothing |
| + h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1") |
| + if err := movesPass(context.Background(), h.s, h.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := h.read(t, host); got.Move.State != provision.MovePreparing { |
| + t.Fatalf("state advanced with no target coords: %+v", got.Move) |
| + } |
| + if len(h.f.ops) != 0 { |
| + t.Fatalf("mirrored toward an unknown target: %v", h.f.ops) |
| + } |
| +} |
| + |
| +func writeFileForTest(path, content string) error { return os.WriteFile(path, []byte(content), 0o600) } |
| +func fileExistsForTest(path string) bool { _, err := os.Stat(path); return err == nil } |
| + |
| +// The rails and the ownership gate (review round 1): pooled, non-hibernate, |
| +// non-instance, and not-my-route records are skipped untouched. |
| +func TestMovesPassRailsAndOwnership(t *testing.T) { |
| + h := newMoveHarness(t) |
| + host := "a.go.example.com" |
| + h.record(t, host, provision.MoveRequest("uk", time.Now())) |
| + h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1") |
| + |
| + // Not this box's route (a standby sharing the region): untouched. |
| + h.owns = false |
| + if err := movesPass(context.Background(), h.s, h.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(h.f.ops) != 0 || h.read(t, host).Move.State != provision.MovePreparing { |
| + t.Fatalf("a non-owning box acted: ops=%v", h.f.ops) |
| + } |
| + h.owns = true |
| + |
| + // A pooled sibling with a move field: skipped (the platform box holds |
| + // none of its data — a round here would promote ready over nothing). |
| + pooled := "p.go.example.com" |
| + rec := provision.Record{Schema: 1, Host: pooled, Slug: "p", App: "app", Account: "bab", |
| + Region: "ie", Pool: "dedicated-x", Channel: "stable", CreatedAt: time.Now().UTC()} |
| + if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", pooled, provision.MoveRequest("uk", time.Now())); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := movesPass(context.Background(), h.s, h.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if h.read(t, pooled).Move.State != provision.MovePreparing { |
| + t.Fatal("a pooled record's move advanced on the platform box") |
| + } |
| +} |
| + |
| +// Abort's box half: staging deleted, then the field cleared. |
| +func TestMovesPassAbortPhase(t *testing.T) { |
| + h := newMoveHarness(t) |
| + host := "a.go.example.com" |
| + h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MoveAbort()) |
| + h.f.put("data-uk", "moves/"+host+"/generations/g/snapshot", "e1") |
| + if err := movesPass(context.Background(), h.s, h.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/") |
| + if len(staged) != 0 { |
| + t.Fatalf("staging survived the abort: %v", staged) |
| + } |
| + if got := h.read(t, host); got.Move != nil || got.Region != "ie" { |
| + t.Fatalf("after abort phase: %+v region=%q", got.Move, got.Region) |
| + } |
| +} |
diff --git a/internal/lease/mirror.go b/internal/lease/mirror.go| new file mode 100644 |
| index 000000000..b0cda7f7d |
| --- /dev/null |
| +++ b/internal/lease/mirror.go |
| @@ -0,0 +1,165 @@ |
| +package lease |
| + |
| +import ( |
| + "context" |
| + "fmt" |
| + "net/url" |
| + "sort" |
| + "strings" |
| + |
| + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| +) |
| + |
| +// The staging mirror (issue #331, instance-move spec): a residence move |
| +// copies the instance's REPLICA prefix — the litestream generations the |
| +// activator already maintains — never the database itself. MirrorPrefix is |
| +// the one delta round; the move worker calls it repeatedly until a round |
| +// reports nothing changed, and once more after the child is down. |
| +// |
| +// The interface exists for the tests (an in-memory fake) and because the |
| +// destination bucket lives in ANOTHER region: the implementation holds one |
| +// client per side. |
| + |
| +// MirrorStore is the three S3 actions a delta round needs. List answers |
| +// key→ETag under a prefix (keys RELATIVE to the bucket root, not the |
| +// prefix); Copy is a server-side CopyObject; Delete removes one key. |
| +type MirrorStore interface { |
| + List(ctx context.Context, bucket, prefix string) (map[string]string, error) |
| + Copy(ctx context.Context, srcBucket, srcKey, dstBucket, dstKey string) error |
| + Delete(ctx context.Context, bucket, key string) error |
| +} |
| + |
| +// MirrorPrefix makes dstBucket/dstPrefix an exact mirror of |
| +// srcBucket/srcPrefix: copy keys that are new or whose ETag differs, delete |
| +// keys the source no longer has (litestream retention must hold in staging |
| +// too — a restore that replays pruned WAL segments is a restore nobody |
| +// tested). Returns how many objects changed; 0 means the sides were |
| +// identical when listed — the move worker's "caught up". |
| +// |
| +// ETag equality is a valid sameness test here even for multipart uploads: |
| +// both sides of a server-side COPY carry the copy's own single-part ETag, |
| +// and litestream's own objects are single PUTs. |
| +func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstBucket, dstPrefix string) (int, error) { |
| + // Both prefixes are DIRECTORY boundaries, normalized here to end in |
| + // "/": activator.HibernatePrefix carries none (path.Join), MovesPrefix |
| + // carries one, and relative-key math over a slashless prefix would both |
| + // mint fused keys (instances/hgenerations/…) and let a listing bleed |
| + // into a sibling host whose name merely extends this one's. |
| + if !strings.HasSuffix(srcPrefix, "/") { |
| + srcPrefix += "/" |
| + } |
| + if !strings.HasSuffix(dstPrefix, "/") { |
| + dstPrefix += "/" |
| + } |
| + src, err := m.List(ctx, srcBucket, srcPrefix) |
| + if err != nil { |
| + return 0, fmt.Errorf("mirror: list source %s/%s: %w", srcBucket, srcPrefix, err) |
| + } |
| + dst, err := m.List(ctx, dstBucket, dstPrefix) |
| + if err != nil { |
| + return 0, fmt.Errorf("mirror: list staging %s/%s: %w", dstBucket, dstPrefix, err) |
| + } |
| + // Deterministic order: a failed round resumes the same way everywhere. |
| + rel := make([]string, 0, len(src)) |
| + for k := range src { |
| + rel = append(rel, k[len(srcPrefix):]) |
| + } |
| + sort.Strings(rel) |
| + changed := 0 |
| + for _, r := range rel { |
| + sk, dk := srcPrefix+r, dstPrefix+r |
| + if et, ok := dst[dk]; ok && et == src[sk] { |
| + continue |
| + } |
| + if err := m.Copy(ctx, srcBucket, sk, dstBucket, dk); err != nil { |
| + return changed, fmt.Errorf("mirror: copy %s: %w", sk, err) |
| + } |
| + changed++ |
| + } |
| + for dk := range dst { |
| + r := dk[len(dstPrefix):] |
| + if _, ok := src[srcPrefix+r]; !ok { |
| + if err := m.Delete(ctx, dstBucket, dk); err != nil { |
| + return changed, fmt.Errorf("mirror: delete %s: %w", dk, err) |
| + } |
| + changed++ |
| + } |
| + } |
| + return changed, nil |
| +} |
| + |
| +// s3Mirror implements MirrorStore over two regional clients: reads and the |
| +// copy call go to the DESTINATION region's client (CopyObject executes in |
| +// the destination bucket's region and pulls from the source), source |
| +// listings to the source's. |
| +type s3Mirror struct { |
| + src, dst *awss3.Client |
| +} |
| + |
| +// NewS3Mirror builds the two-region pair with the same credential chain and |
| +// endpoint compatibility every other S3 consumer here uses (endpoint.go's |
| +// newLeaserClient). |
| +func NewS3Mirror(ctx context.Context, srcRegion, dstRegion string) (MirrorStore, error) { |
| + src, _, err := newLeaserClient(ctx, srcRegion) |
| + if err != nil { |
| + return nil, err |
| + } |
| + dst, _, err := newLeaserClient(ctx, dstRegion) |
| + if err != nil { |
| + return nil, err |
| + } |
| + return &s3Mirror{src: src, dst: dst}, nil |
| +} |
| + |
| +func (m *s3Mirror) client(bucketSide *awss3.Client) *awss3.Client { return bucketSide } |
| + |
| +func (m *s3Mirror) list(ctx context.Context, c *awss3.Client, bucket, prefix string) (map[string]string, error) { |
| + out := map[string]string{} |
| + var token *string |
| + for { |
| + resp, err := c.ListObjectsV2(ctx, &awss3.ListObjectsV2Input{ |
| + Bucket: &bucket, Prefix: &prefix, ContinuationToken: token, |
| + }) |
| + if err != nil { |
| + return nil, err |
| + } |
| + for _, o := range resp.Contents { |
| + if o.Key == nil { |
| + continue |
| + } |
| + et := "" |
| + if o.ETag != nil { |
| + et = *o.ETag |
| + } |
| + out[*o.Key] = et |
| + } |
| + if resp.IsTruncated == nil || !*resp.IsTruncated || resp.NextContinuationToken == nil { |
| + return out, nil |
| + } |
| + token = resp.NextContinuationToken |
| + } |
| +} |
| + |
| +// List tries the destination client first and falls back to the source's — |
| +// the two buckets live in different regions and the caller does not say |
| +// which side a bucket is on. A region-mismatch answer from S3 is an error, |
| +// so the fallback is unambiguous. |
| +func (m *s3Mirror) List(ctx context.Context, bucket, prefix string) (map[string]string, error) { |
| + if out, err := m.list(ctx, m.src, bucket, prefix); err == nil { |
| + return out, nil |
| + } |
| + return m.list(ctx, m.dst, bucket, prefix) |
| +} |
| + |
| +func (m *s3Mirror) Copy(ctx context.Context, srcBucket, srcKey, dstBucket, dstKey string) error { |
| + source := url.PathEscape(srcBucket + "/" + srcKey) |
| + _, err := m.dst.CopyObject(ctx, &awss3.CopyObjectInput{ |
| + Bucket: &dstBucket, Key: &dstKey, CopySource: &source, |
| + }) |
| + return err |
| +} |
| + |
| +func (m *s3Mirror) Delete(ctx context.Context, bucket, key string) error { |
| + _, err := m.dst.DeleteObject(ctx, &awss3.DeleteObjectInput{Bucket: &bucket, Key: &key}) |
| + return err |
| +} |
diff --git a/internal/lease/mirror_test.go b/internal/lease/mirror_test.go| new file mode 100644 |
| index 000000000..3ab8f11b0 |
| --- /dev/null |
| +++ b/internal/lease/mirror_test.go |
| @@ -0,0 +1,112 @@ |
| +package lease |
| + |
| +import ( |
| + "context" |
| + "fmt" |
| + "strings" |
| + "testing" |
| +) |
| + |
| +// fakeMirror is an in-memory MirrorStore: buckets of key→etag, plus an op |
| +// log so ordering tests can assert the sequence. |
| +type fakeMirror struct { |
| + buckets map[string]map[string]string |
| + ops []string |
| + failOn string |
| +} |
| + |
| +func newFakeMirror() *fakeMirror { return &fakeMirror{buckets: map[string]map[string]string{}} } |
| + |
| +func (f *fakeMirror) put(bucket, key, etag string) { |
| + if f.buckets[bucket] == nil { |
| + f.buckets[bucket] = map[string]string{} |
| + } |
| + f.buckets[bucket][key] = etag |
| +} |
| + |
| +func (f *fakeMirror) List(_ context.Context, bucket, prefix string) (map[string]string, error) { |
| + out := map[string]string{} |
| + for k, v := range f.buckets[bucket] { |
| + if strings.HasPrefix(k, prefix) { |
| + out[k] = v |
| + } |
| + } |
| + return out, nil |
| +} |
| + |
| +func (f *fakeMirror) Copy(_ context.Context, srcBucket, srcKey, dstBucket, dstKey string) error { |
| + if f.failOn == srcKey { |
| + return fmt.Errorf("injected copy failure on %s", srcKey) |
| + } |
| + f.ops = append(f.ops, "copy "+srcKey+" -> "+dstBucket+"/"+dstKey) |
| + f.put(dstBucket, dstKey, f.buckets[srcBucket][srcKey]) |
| + return nil |
| +} |
| + |
| +func (f *fakeMirror) Delete(_ context.Context, bucket, key string) error { |
| + f.ops = append(f.ops, "delete "+bucket+"/"+key) |
| + delete(f.buckets[bucket], key) |
| + return nil |
| +} |
| + |
| +func TestMirrorPrefix(t *testing.T) { |
| + ctx := context.Background() |
| + f := newFakeMirror() |
| + f.put("src", "instances/h/generations/a/snapshot", "e1") |
| + f.put("src", "instances/h/generations/a/wal/000", "e2") |
| + |
| + // Initial round: full copy. |
| + n, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| + if err != nil || n != 2 { |
| + t.Fatalf("initial round = %d, %v; want 2 copies", n, err) |
| + } |
| + // Identical: zero. |
| + if n, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err != nil || n != 0 { |
| + t.Fatalf("caught-up round = %d, %v; want 0", n, err) |
| + } |
| + // Delta: one changed etag, one new, one pruned at source. |
| + f.put("src", "instances/h/generations/a/wal/000", "e2b") |
| + f.put("src", "instances/h/generations/a/wal/001", "e3") |
| + delete(f.buckets["src"], "instances/h/generations/a/snapshot") |
| + n, err = MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/") |
| + if err != nil || n != 3 { |
| + t.Fatalf("delta round = %d, %v; want 2 copies + 1 delete", n, err) |
| + } |
| + got, _ := f.List(ctx, "dst", "moves/h/") |
| + want := map[string]string{ |
| + "moves/h/generations/a/wal/000": "e2b", |
| + "moves/h/generations/a/wal/001": "e3", |
| + } |
| + if len(got) != len(want) { |
| + t.Fatalf("staging = %v, want %v", got, want) |
| + } |
| + for k, v := range want { |
| + if got[k] != v { |
| + t.Fatalf("staging[%s] = %q, want %q", k, got[k], v) |
| + } |
| + } |
| + // A copy failure surfaces with the partial count. |
| + f.put("src", "instances/h/generations/a/wal/002", "e4") |
| + f.failOn = "instances/h/generations/a/wal/002" |
| + if _, err := MirrorPrefix(ctx, f, "src", "instances/h/", "dst", "moves/h/"); err == nil { |
| + t.Fatal("injected failure did not surface") |
| + } |
| +} |
| + |
| +// Prefix boundaries: a slashless prefix (activator.HibernatePrefix's shape) |
| +// is normalized to a directory boundary — no fused keys, and a sibling host |
| +// whose name extends this one's is never swept in. |
| +func TestMirrorPrefixNormalizesBoundaries(t *testing.T) { |
| + ctx := context.Background() |
| + f := newFakeMirror() |
| + f.put("src", "instances/h/generations/a/snapshot", "e1") |
| + f.put("src", "instances/h2/generations/a/snapshot", "eX") // the sibling |
| + n, err := MirrorPrefix(ctx, f, "src", "instances/h", "dst", "moves/h") |
| + if err != nil || n != 1 { |
| + t.Fatalf("round = %d, %v; want exactly the one h object", n, err) |
| + } |
| + got, _ := f.List(ctx, "dst", "") |
| + if _, ok := got["moves/h/generations/a/snapshot"]; !ok || len(got) != 1 { |
| + t.Fatalf("staging keys = %v, want the one properly-joined key", got) |
| + } |
| +} |
diff --git a/internal/provision/provision.go b/internal/provision/provision.go| index 2108074b1..05acf3438 100644 |
| --- a/internal/provision/provision.go |
| +++ b/internal/provision/provision.go |
| @@ -42,6 +42,12 @@ type Record struct { |
| CreatedAt time.Time `json:"created_at"` |
| CreatedBy string `json:"created_by,omitempty"` |
| + // Move is an in-flight residence move (issue #331, instance-move spec): |
| + // present only between move-prepare and the cut's terminal flip, absent |
| + // on every record at rest. Additive + omitempty — pre-move records parse |
| + // unchanged. Written ONLY through SetRecordMove's guarded transitions. |
| + Move *MoveState `json:"move,omitempty"` |
| + |
| // --- shape fields (2026-08-10 route-records design §3) -------------- |
| // Every one is omitempty with a zero value that reproduces the mint |
| // reconcile hardcoded before they existed: a record written by an older |
| @@ -671,6 +677,165 @@ func SetRecordRegion(ctx context.Context, s release.Store, acct, app, host, regi |
| return nil |
| } |
| +// MoveState is the record's in-flight move (issue #331). State is one of |
| +// the Move* constants; To is the target placement code. ReadyAt is stamped |
| +// by the source box's preparing→ready promotion, so the tenant can see how |
| +// stale its freeze-window budget is. |
| +type MoveState struct { |
| + To string `json:"to"` |
| + State string `json:"state"` |
| + RequestedAt time.Time `json:"requested_at"` |
| + ReadyAt *time.Time `json:"ready_at,omitempty"` |
| +} |
| + |
| +// The move states, in order. There is no "done" state: the terminal |
| +// transition flips Region and CLEARS Move in one write — a record at rest |
| +// never says anything about moves. |
| +const ( |
| + MovePreparing = "preparing" |
| + MoveReady = "ready" |
| + MoveCutting = "cutting" |
| + // MoveAborting is abort's in-flight half (review round 1): the console |
| + // marks it, the SOURCE box deletes the staged copy from the target |
| + // bucket and then clears Move — because a staged replica left behind by |
| + // a bare clear would win a LATER landing on that box (staging-wins, |
| + // backend_exec.go's promoteStagedReplica) with data from an abandoned |
| + // move. |
| + MoveAborting = "aborting" |
| +) |
| + |
| +// ErrMoveConflict is a transition refused because the record's current move |
| +// state does not admit it — a second prepare while one is in flight, a cut |
| +// before ready, an abort after the point of no return. The console maps it |
| +// to 409; the box treats it as "someone else already advanced this" and |
| +// re-reads. |
| +var ErrMoveConflict = errors.New("provision: move state conflict") |
| + |
| +// SetRecordMove applies one guarded move transition: the same narrow |
| +// read-CAS-write with one retry that SetRecordRegion uses (and for the same |
| +// reason — re-reading and re-applying a transition on a lost race changes |
| +// nothing about what it meant; an ILLEGAL transition after the re-read is |
| +// ErrMoveConflict, which is the answer, not a race). mut sees the freshly |
| +// read record and either mutates it or returns the error that names why it |
| +// can't. |
| +func SetRecordMove(ctx context.Context, s release.Store, acct, app, host string, mut func(*Record) error) error { |
| + key := RecordKey(acct, app, host) |
| + update := func() error { |
| + raw, etag, err := s.Get(ctx, key) |
| + if err != nil { |
| + return err |
| + } |
| + var rec Record |
| + if err := json.Unmarshal(raw, &rec); err != nil { |
| + return fmt.Errorf("provision: record %s/%s/%s: %w", acct, app, host, err) |
| + } |
| + if err := mut(&rec); err != nil { |
| + return err |
| + } |
| + body, err := json.Marshal(rec) |
| + if err != nil { |
| + return err |
| + } |
| + _, err = s.Put(ctx, key, body, etag) |
| + return err |
| + } |
| + if err := update(); errors.Is(err, release.ErrCASConflict) { |
| + return update() // one retry on a lost race; a second loss surfaces |
| + } else if err != nil { |
| + return err |
| + } |
| + return nil |
| +} |
| + |
| +// The five transitions, each a mut for SetRecordMove. They are the ONLY |
| +// writers of Record.Move. |
| + |
| +// MoveRequest starts a move to region `to`. Refused when one is already in |
| +// flight — the caller aborts first or waits. |
| +func MoveRequest(to string, now time.Time) func(*Record) error { |
| + return func(rec *Record) error { |
| + if rec.Move != nil { |
| + return fmt.Errorf("%w: a move to %s is already %s", ErrMoveConflict, rec.Move.To, rec.Move.State) |
| + } |
| + if rec.Region == to { |
| + return fmt.Errorf("%w: the instance is already in %s", ErrMoveConflict, to) |
| + } |
| + rec.Move = &MoveState{To: to, State: MovePreparing, RequestedAt: now.UTC()} |
| + return nil |
| + } |
| +} |
| + |
| +// MovePromoteReady is the source box's preparing→ready promotion. |
| +func MovePromoteReady(now time.Time) func(*Record) error { |
| + return func(rec *Record) error { |
| + if rec.Move == nil || rec.Move.State != MovePreparing { |
| + return fmt.Errorf("%w: ready is only reached from preparing", ErrMoveConflict) |
| + } |
| + t := now.UTC() |
| + rec.Move.State = MoveReady |
| + rec.Move.ReadyAt = &t |
| + return nil |
| + } |
| +} |
| + |
| +// MoveMarkCutting is the console's cut verb: legal only from ready — the |
| +// tenant freezes writes first, and a cut from preparing would cut over a |
| +// replica that was never caught up. |
| +func MoveMarkCutting() func(*Record) error { |
| + return func(rec *Record) error { |
| + if rec.Move == nil || rec.Move.State != MoveReady { |
| + return fmt.Errorf("%w: cut is only legal from ready", ErrMoveConflict) |
| + } |
| + rec.Move.State = MoveCutting |
| + return nil |
| + } |
| +} |
| + |
| +// MoveAbort marks a move for abort — legal before the point of no return. |
| +// The clear itself is the source box's (MoveAbortComplete), after it has |
| +// deleted the staged copy. |
| +func MoveAbort() func(*Record) error { |
| + return func(rec *Record) error { |
| + if rec.Move == nil { |
| + return fmt.Errorf("%w: no move in flight", ErrMoveConflict) |
| + } |
| + switch rec.Move.State { |
| + case MoveCutting: |
| + return fmt.Errorf("%w: past the point of no return — the cut completes forward", ErrMoveConflict) |
| + case MoveAborting: |
| + return fmt.Errorf("%w: already aborting", ErrMoveConflict) |
| + } |
| + rec.Move.State = MoveAborting |
| + return nil |
| + } |
| +} |
| + |
| +// MoveAbortComplete is the source box's half of an abort: staging is gone, |
| +// the move field clears. |
| +func MoveAbortComplete() func(*Record) error { |
| + return func(rec *Record) error { |
| + if rec.Move == nil || rec.Move.State != MoveAborting { |
| + return fmt.Errorf("%w: abort-complete is only legal from aborting", ErrMoveConflict) |
| + } |
| + rec.Move = nil |
| + return nil |
| + } |
| +} |
| + |
| +// MoveComplete is the source's terminal act: Region flips to the target and |
| +// Move clears, in ONE write — every reader that knows nothing about moves |
| +// sees the old region right up until it sees the new one. |
| +func MoveComplete() func(*Record) error { |
| + return func(rec *Record) error { |
| + if rec.Move == nil || rec.Move.State != MoveCutting { |
| + return fmt.Errorf("%w: complete is only legal from cutting", ErrMoveConflict) |
| + } |
| + rec.Region = rec.Move.To |
| + rec.Move = nil |
| + return nil |
| + } |
| +} |
| + |
| // DeleteRecord retires an instance's route authorization. Object deletion |
| // only — the parked database and blobs under the regional bucket's prefix |
| // are deliberately untouched (design §3.2); Store's Delete is idempotent, |
diff --git a/internal/provision/provision_test.go b/internal/provision/provision_test.go| index c92266f39..a152520b3 100644 |
| --- a/internal/provision/provision_test.go |
| +++ b/internal/provision/provision_test.go |
| @@ -886,3 +886,111 @@ func TestHealthWireTagIsProbeNotHealth(t *testing.T) { |
| t.Fatalf("marshal used the collided health key: %s", b) |
| } |
| } |
| + |
| +// The move transitions (issue #331): SetRecordMove is the one writer of |
| +// Record.Move, and each transition validates the state it leaves from. The |
| +// terminal MoveComplete flips Region and clears Move in ONE write. |
| +func TestSetRecordMoveTransitions(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := Record{Schema: 1, Host: "a.go.example.com", Slug: "a", App: "app", Account: "bab", |
| + Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()} |
| + if err := CreateRecord(ctx, s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + mv := func(mut func(*Record) error) error { |
| + return SetRecordMove(ctx, s, "bab", "app", "a.go.example.com", mut) |
| + } |
| + read := func() Record { |
| + t.Helper() |
| + got, err := GetRecord(ctx, s, "bab", "app", "a.go.example.com") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return *got |
| + } |
| + now := time.Now() |
| + |
| + if err := mv(MoveRequest("ie", now)); !errors.Is(err, ErrMoveConflict) { |
| + t.Fatalf("move to the current region = %v, want ErrMoveConflict", err) |
| + } |
| + if err := mv(MoveMarkCutting()); !errors.Is(err, ErrMoveConflict) { |
| + t.Fatalf("cut with no move in flight = %v, want ErrMoveConflict", err) |
| + } |
| + if err := mv(MoveRequest("uk", now)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := read(); got.Move == nil || got.Move.State != MovePreparing || got.Move.To != "uk" || got.Region != "ie" { |
| + t.Fatalf("after request: %+v", got.Move) |
| + } |
| + if err := mv(MoveRequest("se", now)); !errors.Is(err, ErrMoveConflict) { |
| + t.Fatalf("second request while in flight = %v, want ErrMoveConflict", err) |
| + } |
| + if err := mv(MoveMarkCutting()); !errors.Is(err, ErrMoveConflict) { |
| + t.Fatalf("cut from preparing = %v, want ErrMoveConflict", err) |
| + } |
| + if err := mv(MovePromoteReady(now)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := read(); got.Move.State != MoveReady || got.Move.ReadyAt == nil { |
| + t.Fatalf("after promote: %+v", got.Move) |
| + } |
| + if err := mv(MoveMarkCutting()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := mv(MoveAbort()); !errors.Is(err, ErrMoveConflict) { |
| + t.Fatalf("abort from cutting = %v, want ErrMoveConflict (completes forward)", err) |
| + } |
| + if err := mv(MoveComplete()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := read(); got.Region != "uk" || got.Move != nil { |
| + t.Fatalf("after complete: region=%q move=%+v, want uk/nil in one write", got.Region, got.Move) |
| + } |
| + // Abort is two-phase: ready → aborting (cut and promote both refuse |
| + // there) → cleared only by MoveAbortComplete. |
| + if err := mv(MoveRequest("ie", now)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := mv(MovePromoteReady(now)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := mv(MoveAbort()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := read(); got.Move == nil || got.Move.State != MoveAborting { |
| + t.Fatalf("after abort: %+v, want aborting", got.Move) |
| + } |
| + if err := mv(MoveMarkCutting()); !errors.Is(err, ErrMoveConflict) { |
| + t.Fatalf("cut from aborting = %v, want ErrMoveConflict", err) |
| + } |
| + if err := mv(MoveAbort()); !errors.Is(err, ErrMoveConflict) { |
| + t.Fatalf("double abort = %v, want ErrMoveConflict", err) |
| + } |
| + if err := mv(MoveAbortComplete()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := read(); got.Move != nil || got.Region != "uk" { |
| + t.Fatalf("after abort-complete: %+v region=%q", got.Move, got.Region) |
| + } |
| +} |
| + |
| +// A record written before the move field existed round-trips byte-identical |
| +// through the CAS helper's read-marshal cycle — the wire-shape rule. |
| +func TestMoveFieldAbsentOnPreMoveRecords(t *testing.T) { |
| + raw := []byte(`{"schema":1,"host":"a.go.example.com","slug":"a","app":"app","account":"bab","region":"ie","channel":"stable","created_at":"2026-08-01T00:00:00Z"}`) |
| + var rec Record |
| + if err := json.Unmarshal(raw, &rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + if rec.Move != nil { |
| + t.Fatalf("pre-move record decoded a Move: %+v", rec.Move) |
| + } |
| + out, err := json.Marshal(rec) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if strings.Contains(string(out), "move") { |
| + t.Fatalf("re-marshal grew a move key: %s", out) |
| + } |
| +} |
diff --git a/internal/registry/registry.go b/internal/registry/registry.go| index 0c9a5fafa..130626a37 100644 |
| --- a/internal/registry/registry.go |
| +++ b/internal/registry/registry.go |
| @@ -832,6 +832,27 @@ func (r *Registry) SetChannel(host, channel string) error { |
| return nil |
| } |
| +// SetStatus writes a route's lifecycle status word. Same contract as |
| +// SetChannel: a narrow UPDATE that cannot clobber concurrent column writes, |
| +// and an error when no row matched. The residence-move cut (internal/host's |
| +// move worker) is its first caller — status "moving" takes a route out of |
| +// the edge's served set and the activator's managed set in one write, which |
| +// is what quiesces the child ahead of the final mirror round. |
| +func (r *Registry) SetStatus(host, status string) error { |
| + res, err := r.sql.Exec(`UPDATE routes SET status = ? WHERE host = ?`, status, host) |
| + if err != nil { |
| + return err |
| + } |
| + n, err := res.RowsAffected() |
| + if err != nil { |
| + return err |
| + } |
| + if n == 0 { |
| + return fmt.Errorf("no route for host %q", host) |
| + } |
| + return nil |
| +} |
| + |
| // SetCaps writes a route's operator-granted capability tokens — the ONLY |
| // writer of that column, and deliberately the only one: `carlos route |
| // -grant/-revoke` is the operator's door to a transfer of box authority, and |