0a6bb375c · carlos/platform

diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go

index 3eac4dc67..7d15e7bba 100644

--- a/approved_moves_copy_test.go

+++ b/approved_moves_copy_test.go

@@ -107,6 +107,7 @@ var movesCopyIDs = []string{

"MV-O172", "MV-O173", "MV-O174", "MV-O175", "MV-O176", "MV-O177",

"MV-O178",

"MV-O179", "MV-O180", "MV-O181", "MV-O182", "MV-O183", "MV-O184", "MV-O185", "MV-O186", "MV-O187", "MV-O188", "MV-O189",

+ "MV-O190", "MV-O191", "MV-O192",

}

// copyRow is one row of the record. The File cell is the file the sentence is

diff --git a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md

index 82873e003..0d785b5b0 100644

--- a/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md

+++ b/docs/superpowers/plans/2026-10-06-moves-phase1-copy.md

@@ -709,3 +709,11 @@ MV-O181 is the usage line with `stage` added. It replaces MV-O56, whose withdraw

| MV-O189 | docs/site/cli.md | `stage` is optional and comes before `start`. It copies the instance's data to the region it moves to while the instance keeps serving, and each time it runs it copies only what is new. The more it has copied, the less is left to copy while the instance is offline. Until `start`, `abort` removes the staged copy and the instance is never offline. | approved |

+

+Three more console door refusals for a staged move (`internal/console/apiinstancemove_offline.go`, Astra round 2), pending Luna's review. MV-O190 is what `start` and `seal` answer when the box that staged the instance (the stager, pinned on the record at the stage) no longer publishes an edges file in the source region: its absence is not proof that it stopped copying, so the step waits for it or the operator aborts. `<box>` is the stager, `<from>` the source region's name. MV-O191 is the stage door's refusal of a target whose data bucket is a source box's own, where a staged copy would replace the serving replica. MV-O192 is what `stage`, `start` and `seal` answer once the target publishes another data bucket or region than the one the stage fixed on the record: the staged copy is in the stored bucket and is never followed elsewhere.

+

+|---|---|---|---|

diff --git a/internal/console/apiinstancemove_offline.go b/internal/console/apiinstancemove_offline.go

index c8fe845a5..b8097a663 100644

--- a/internal/console/apiinstancemove_offline.go

+++ b/internal/console/apiinstancemove_offline.go

@@ -567,12 +567,49 @@ func offlineStaged(rec *provision.Record) (*provision.MoveState, bool) {

return nil, false

}

+// offlineStagerGone is MV-O190: the pinned stager of a staged move no

+// longer publishes an edges file in the source region. Its disappearance

+// is never proof that it stopped copying, so start and seal refuse.

+func (s *Server) offlineStagerGone(box, host, from string) *offlineRefusal {

+ return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s, which staged %s, no longer publishes an edges file in %s. bring it back or abort the move", box, host, from)}

+}

+

+// offlineStageCoords refuses a staged move whose target publishes other

+// staging coordinates than the ones the stage fixed on the record (MV-O192,

+// Astra round 2, finding 2): the staged copy is in the stored bucket, and

+// nothing is ever retargeted. offlineSelect has already required exactly

+// one target box.

+func (s *Server) offlineStageCoords(f offlineFleet, mv *provision.MoveState, host string) *offlineRefusal {

+ if mv.StagedAt == nil {

+ return nil

+ }

+ tgt := f.boxesAt(mv.To)

+ if len(tgt) != 1 || tgt[0].DataBucket != mv.StageBucket || tgt[0].DataBucketRegion != mv.StageRegion {

+ return &offlineRefusal{status: http.StatusConflict, msg: fmt.Sprintf("%s now publishes a different data bucket than the one %s was staged in. abort the move and stage it again", s.offlineName(mv.To), host)}

+ }

+ return nil

+}

+

+// offlineStagerPresent refuses a staged move whose pinned stager is not in

+// the source region's fleet now (MV-O190).

+func (s *Server) offlineStagerPresent(f offlineFleet, mv *provision.MoveState, host string) *offlineRefusal {

+ if mv.StagedAt == nil {

+ return nil

+ }

+ for _, b := range f.boxesAt(mv.From) {

+ if b.Box == mv.Stager {

+ return nil

+ }

+ return s.offlineStagerGone(mv.Stager, host, s.offlineName(mv.From))

+}

+

func (s *Server) handleAPIInstanceOfflineStart(w http.ResponseWriter, r *http.Request) {

d, ok := s.offlineGate(w, r)

if !ok {

return

}

- to, _, ok := s.offlineBeginChecks(w, r, d)

+ to, fleet, ok := s.offlineBeginChecks(w, r, d)

if !ok {

return

}

@@ -583,6 +620,14 @@ func (s *Server) handleAPIInstanceOfflineStart(w http.ResponseWriter, r *http.Re

writeRefusal(w, s.offlineStagedFor(d.host, mv.To))

return

}

+ if ref := s.offlineStagerPresent(fleet, mv, d.host); ref != nil {

+ writeRefusal(w, ref)

+ return

+ }

+ if ref := s.offlineStageCoords(fleet, mv, d.host); ref != nil {

+ writeRefusal(w, ref)

+ return

+ }

s.offlineAdvance(w, r, d, provision.OfflineStartStaged(mv.ID, d.addr, time.Now()))

return

}

@@ -611,6 +656,10 @@ func (s *Server) handleAPIInstanceOfflineStage(w http.ResponseWriter, r *http.Re

writeRefusal(w, s.offlineStagedFor(d.host, mv.To))

return

}

+ if ref := s.offlineStageCoords(fleet, mv, d.host); ref != nil {

+ writeRefusal(w, ref)

+ return

+ }

s.offlineAdvance(w, r, d, provision.OfflineRestage(mv.ID, time.Now()))

return

}

@@ -628,9 +677,21 @@ func (s *Server) handleAPIInstanceOfflineStage(w http.ResponseWriter, r *http.Re

for _, b := range fleet.boxesAt(d.rec.Region) {

names = append(names, b.Box)

}

- stager := moves.OfflineStager(names)

+ // The staging destination is fixed here too, before any copy (Astra

+ // round 2, finding 2): the one target box's data bucket and region, as

+ // offlineSelect found it. A target that publishes a source box's own

+ // bucket is refused: the activator there would promote the staged copy

+ // over the serving replica on its next wake (MV-O191).

+ tgt := fleet.boxesAt(to)[0]

+ for _, b := range fleet.boxesAt(d.rec.Region) {

+ if b.DataBucket == tgt.DataBucket {

+ jsonError(w, http.StatusConflict, fmt.Sprintf("%s publishes the same data bucket as %s. a staged copy there would replace the serving one", s.offlineName(to), s.offlineName(d.rec.Region)))

+ return

+ }

+ st := provision.StageTarget{Stager: moves.OfflineStager(names), Bucket: tgt.DataBucket, Region: tgt.DataBucketRegion}

s.offlineBegin(w, r, d, func(id string) func(*provision.Record) error {

- return provision.OfflineStage(id, d.rec.Region, to, d.addr, stager, time.Now())

+ return provision.OfflineStage(id, d.rec.Region, to, d.addr, st, time.Now())

})

}

@@ -858,6 +919,12 @@ func (s *Server) offlineStepFor(ctx context.Context, d offlineDoor, mv *provisio

if ref := s.offlineSelect(ctx, fleet, host, mv.From, mv.To); ref != nil {

return offlineStep{}, ref

}

+ if ref := s.offlineStagerPresent(fleet, mv, host); ref != nil {

+ return offlineStep{}, ref

+ }

+ if ref := s.offlineStageCoords(fleet, mv, host); ref != nil {

+ return offlineStep{}, ref

+ }

return offlineStep{mut: provision.OfflineStartStaged(mv.ID, d.addr, now)}, nil

case provision.OfflineStopping:

fleet, ok := s.readOfflineFleet(ctx)

@@ -929,6 +996,28 @@ func (s *Server) offlineSealCheck(f offlineFleet, ev offlineEvidence, mv *provis

if len(src) == 0 {

return "", nil, &offlineRefusal{status: http.StatusBadRequest, msg: "no eligible box in " + from + " publishes an edges file"}

}

+ // A staged move's pinned stager is part of the barrier whatever the

+ // fleet says (Astra round 2, finding 1): it must still be in the

+ // source region's fleet (so it is one of the boxes below, and of the

+ // boxes the seal names), and it must have said, after the move left

+ // staging and under its pass lock, that no stage round is in flight

+ // and none will start (its stager report, phase stopped, fresh). The

+ // staging destination must still be the one the stage fixed.

+ if mv.StagedAt != nil {

+ if ref := s.offlineStagerPresent(f, mv, host); ref != nil {

+ return "", nil, ref

+ }

+ if ref := s.offlineStageCoords(f, mv, host); ref != nil {

+ return "", nil, ref

+ }

+ ack, ok := ev.get(moves.OfflineRoleStager, mv.Stager)

+ if ok && ack.Phase == moves.OfflinePhaseStopped && !offlineFresh(ack.OfflineReport, now) {

+ return "", nil, offlineStaleRefusal(mv.Stager)

+ }

+ if !ok || ack.Phase != moves.OfflinePhaseStopped {

+ return "", nil, s.offlineWaitRefusal(mv.Stager, from, host)

+ }

var data []string

for _, b := range src {

rep, ok := ev.get(moves.OfflineRoleSource, b.Box)

diff --git a/internal/console/apiinstancemove_offline_stage_test.go b/internal/console/apiinstancemove_offline_stage_test.go

index 470342689..b598a9cfb 100644

--- a/internal/console/apiinstancemove_offline_stage_test.go

+++ b/internal/console/apiinstancemove_offline_stage_test.go

@@ -1,7 +1,9 @@

package console

import (

+ "context"

"net/http"

+ "slices"

"testing"

"time"

@@ -118,6 +120,7 @@ func TestOfflineStageThenStartThenSeal(t *testing.T) {

// again, and seal follows the stop.

w.refused(http.MethodPost, "/stage", `{"region":"uk"}`, http.StatusConflict, "a move to uk is already "+provision.OfflineStopping)

w.stopped(nil, "ie-box")

+ w.stagerStopped("ie-box", time.Now())

w.ok("/seal", "", provision.OfflineSealing)

}

@@ -206,3 +209,117 @@ func TestOfflineStagePinsTheStager(t *testing.T) {

t.Fatalf("stager %q (view %q), want ie-a", rec.Move.Stager, v.Stager)

}

+

+// ---- Astra round 2 ----

+

+func msgO190(box string) string {

+ return box + ", which staged " + offlineHost + ", no longer publishes an edges file in ie. bring it back or abort the move"

+}

+

+const (

+ msgO191 = "uk publishes the same data bucket as ie. a staged copy there would replace the serving one"

+ msgO192 = "uk now publishes a different data bucket than the one " + offlineHost + " was staged in. abort the move and stage it again"

+)

+

+// unpublish removes a box from the fleet's evidence, its edges file and

+// its status report: the box left the fleet (it may still be running).

+func (w *offlineWorld) unpublish(box string) {

+ w.t.Helper()

+ for _, key := range []string{edgesFilePrefix + box + ".json", "status/boxes/" + box + ".json"} {

+ if err := w.fx.rel.Delete(context.Background(), key); err != nil {

+ w.t.Fatal(err)

+ }

+}

+

+// stagerStopped writes the pinned stager's stop acknowledgement.

+func (w *offlineWorld) stagerStopped(box string, at time.Time) {

+ w.t.Helper()

+ w.report(moves.OfflineRoleStager, box, moves.OfflinePhaseStopped, func(r *moves.OfflineReport) {

+ r.Held, r.Fenced, r.Quiesced, r.At = false, false, false, at

+ })

+}

+

+// Finding 1: seal waits for the pinned stager's fresh acknowledgement that

+// no round is in flight and none will start, beside every box's stop, and

+// names the stager among the boxes it captures.

+func TestOfflineSealWaitsForThePinnedStager(t *testing.T) {

+ w := newOfflineWorld(t, ieOff("ie-a"), ieOff("ie-b"), ukOff("uk-box"))

+ w.stage()

+ w.ok("", `{"region":"uk"}`, provision.OfflineStopping)

+ w.stopped(nil, "ie-a", "ie-b")

+ w.refused(http.MethodPost, "/seal", "", http.StatusConflict, "Waiting for ie-a in ie to confirm it has stopped "+offlineHost+". "+offlineHost+" is offline.")

+ w.stagerStopped("ie-a", time.Now().Add(-10*time.Minute))

+ w.refused(http.MethodPost, "/seal", "", http.StatusConflict, msgO178("ie-a"))

+ w.stagerStopped("ie-a", time.Now())

+ w.ok("/seal", "", provision.OfflineSealing)

+ if rec := w.record(); !slices.Contains(rec.Move.Boxes, "ie-a") {

+ t.Fatalf("the seal's boxes %v leave out the stager", rec.Move.Boxes)

+ }

+}

+

+// Finding 1: a pinned stager that no longer publishes an edges file is never

+// taken to have stopped. start and seal refuse with MV-O190.

+func TestOfflineStagedMoveRefusesAMissingStager(t *testing.T) {

+ w := newOfflineWorld(t, ieOff("ie-a"), ieOff("ie-b"), ukOff("uk-box"))

+ w.stage()

+ w.unpublish("ie-a")

+ w.refused(http.MethodPost, "", `{"region":"uk"}`, http.StatusConflict, msgO190("ie-a"))

+ if st := w.status(); st.Next != "" || st.Waiting != msgO190("ie-a") {

+ t.Fatalf("status next %q waiting %q", st.Next, st.Waiting)

+ }

+

+ w2 := newOfflineWorld(t, ieOff("ie-a"), ieOff("ie-b"), ukOff("uk-box"))

+ w2.stage()

+ w2.ok("", `{"region":"uk"}`, provision.OfflineStopping)

+ w2.stopped(nil, "ie-a", "ie-b")

+ w2.stagerStopped("ie-a", time.Now())

+ w2.unpublish("ie-a")

+ w2.refused(http.MethodPost, "/seal", "", http.StatusConflict, msgO190("ie-a"))

+}

+

+// Finding 2: the stage fixes its destination on the record, and refuses a

+// target that publishes a source box's own bucket.

+func TestOfflineStageFixesItsDestination(t *testing.T) {

+ w := newOfflineWorld(t, oneEach...)

+ w.stage()

+ if m := w.record().Move; m.StageBucket != "data-uk" || m.StageRegion != "eu-west-2" {

+ t.Fatalf("staging destination %q %q", m.StageBucket, m.StageRegion)

+ }

+ w2 := newOfflineWorld(t, ieOff("ie-box"), offlineBox{Box: "uk-box", Placement: "uk", Bucket: "data-ie"})

+ w2.refused(http.MethodPost, "/stage", `{"region":"uk"}`, http.StatusConflict, msgO191)

+}

+

+// Finding 2: once staged, a target that publishes other coordinates is

+// refused by stage, start and seal, never followed.

+func TestOfflineStagedMoveRefusesChangedCoordinates(t *testing.T) {

+ for _, c := range []struct {

+ name string

+ target offlineBox

+ }{

+ {"another bucket", offlineBox{Box: "uk-box", Placement: "uk", Bucket: "data-uk-new"}},

+ {"another box", offlineBox{Box: "uk-new", Placement: "uk", Bucket: "data-uk-new"}},

+ } {

+ t.Run(c.name, func(t *testing.T) {

+ w := newOfflineWorld(t, oneEach...)

+ w.stage()

+ if c.target.Box != "uk-box" {

+ w.unpublish("uk-box")

+ }

+ w.fleet(c.target)

+ w.refused(http.MethodPost, "/stage", `{"region":"uk"}`, http.StatusConflict, msgO192)

+ w.refused(http.MethodPost, "", `{"region":"uk"}`, http.StatusConflict, msgO192)

+

+ w2 := newOfflineWorld(t, oneEach...)

+ w2.stage()

+ w2.ok("", `{"region":"uk"}`, provision.OfflineStopping)

+ w2.stopped(nil, "ie-box")

+ w2.stagerStopped("ie-box", time.Now())

+ if c.target.Box != "uk-box" {

+ w2.unpublish("uk-box")

+ }

+ w2.fleet(c.target)

+ w2.refused(http.MethodPost, "/seal", "", http.StatusConflict, msgO192)

+ })

+ }

+}

diff --git a/internal/console/apiinstancemove_offline_test.go b/internal/console/apiinstancemove_offline_test.go

index caa67eb91..7b6d0c144 100644

--- a/internal/console/apiinstancemove_offline_test.go

+++ b/internal/console/apiinstancemove_offline_test.go

@@ -157,7 +157,7 @@ func (w *offlineWorld) fleet(boxes ...offlineBox) {

}

body, err := json.Marshal(map[string]any{

"v": 1, "box": b.Box, "box_id": b.Box, "eip": "1.2.3.4", "region": aws, "placement": b.Placement,

- "data_bucket": b.Bucket, "hibernation_configured": !b.NoHibernation, "serves_shared_pool": !b.NotSharedPool,

+ "data_bucket": b.Bucket, "data_bucket_region": aws, "hibernation_configured": !b.NoHibernation, "serves_shared_pool": !b.NotSharedPool,

"accepts_inbound": true, "secrets_key_fp": fp, "capabilities": caps, "hosts": hosts,

})

if err != nil {

diff --git a/internal/edgesfile/snapshot.go b/internal/edgesfile/snapshot.go

index cdaf1333a..bbfd5c6b5 100644

--- a/internal/edgesfile/snapshot.go

+++ b/internal/edgesfile/snapshot.go

@@ -37,6 +37,9 @@ type PlacementView struct {

// coordinates, and doubles as positive hosting evidence from a writer

// too old to publish HibernationConfigured.

DataBucket string `json:"data_bucket"`

+ // DataBucketRegion is that bucket's AWS region, as the box publishes it

+ // beside DataBucket. An offline move's stage fixes both on the record.

+ DataBucketRegion string `json:"data_bucket_region"`

// HibernationConfigured is the box's own answer to "is my hibernation

// activator wired" (internal/host/edgesfile.go). A POINTER because

// absent and false are different facts and the difference is the whole

diff --git a/internal/host/movesoffline.go b/internal/host/movesoffline.go

index 75bc925df..a78e946af 100644

--- a/internal/host/movesoffline.go

+++ b/internal/host/movesoffline.go

@@ -543,7 +543,9 @@ func (w *movesWorker) offlineSourceStep(ctx context.Context, s release.Store, re

}

switch rec.Move.State {

case provision.OfflineStopping, provision.OfflineSealing, provision.OfflineSealed, provision.OfflineLanding, provision.OfflineLanded:

- return w.offlineSourceStop(ctx, s, rec)

+ // The pinned stager's stop acknowledgement first: the seal door

+ // needs it beside this box's stop (movesoffline_stage.go).

+ return errors.Join(w.offlineStagerStopped(ctx, s, rec), w.offlineSourceStop(ctx, s, rec))

case provision.OfflineOpen:

return w.offlineSourcePark(ctx, s, rec)

case provision.OfflineFinishing:

diff --git a/internal/host/movesoffline_seal.go b/internal/host/movesoffline_seal.go

index a0056e640..f72f38d61 100644

--- a/internal/host/movesoffline_seal.go

+++ b/internal/host/movesoffline_seal.go

@@ -351,6 +351,12 @@ func (w *movesWorker) offlineTakeSeal(ctx context.Context, s release.Store, rec

if err != nil {

return seal, err

}

+ // A staged move hands over in the destination its stage fixed (Astra

+ // round 2, finding 2). A target that publishes anything else now is

+ // not sealed into; the seal door refuses it first (MV-O192).

+ if m.StagedAt != nil && (tgt.DataBucket != m.StageBucket || tgt.DataBucketRegion != m.StageRegion) {

+ return seal, fmt.Errorf("the target now publishes %s (%s), not the staging bucket %s (%s) the stage fixed", tgt.DataBucket, tgt.DataBucketRegion, m.StageBucket, m.StageRegion)

+ }

mirror, err := w.offlineMirror(ctx, tgt.DataBucketRegion)

if err != nil {

return seal, err

diff --git a/internal/host/movesoffline_stage.go b/internal/host/movesoffline_stage.go

index ef934d921..53690e6ad 100644

--- a/internal/host/movesoffline_stage.go

+++ b/internal/host/movesoffline_stage.go

@@ -8,6 +8,7 @@ import (

"log"

"os"

"strings"

+ "sync"

"time"

"github.com/carlosframework/platform/internal/activator"

@@ -68,6 +69,17 @@ func (w *movesWorker) offlineStageStep(ctx context.Context, s release.Store, rec

if m.Stager == "" || m.Stager != cfg.Box {

return nil

}

+ // The record again, now, under this box's pass lock: the pass's listing

+ // may predate the start, and once the move has left staging no round

+ // ever starts (Astra round 2, finding 1).

+ cur, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host)

+ if err != nil {

+ return fmt.Errorf("offline move %s: read the record before a stage round: %w", m.ID, err)

+ }

+ if cur.Move == nil || cur.Move.ID != m.ID || cur.Move.State != provision.OfflineStaging {

+ return nil

+ }

+ rec, m = *cur, cur.Move

var request time.Time

if m.StagedAt != nil {

request = m.StagedAt.UTC()

@@ -112,32 +124,54 @@ func (p stageProgress) fill(r *moves.OfflineReport) {

}

// offlineStageRound is one pre-copy round and the listing that measures it.

+// It writes only to the staging destination the stage fixed on the record

+// (Astra round 2, finding 2), and only while the target still publishes

+// exactly that bucket and region; a target that publishes anything else is

+// refused, never followed. The round's delete phase reads the record again

+// first and stops if the move has left staging (finding 1).

func (w *movesWorker) offlineStageRound(ctx context.Context, s release.Store, rec provision.Record) (stageProgress, error) {

cfg := w.cfg

- host := rec.Host

- tgt, err := w.offlineTarget(ctx, s, rec.Move)

+ host, m := rec.Host, rec.Move

+ if m.StageBucket == "" {

+ return stageProgress{}, errors.New("the record names no staging bucket")

+ }

+ tgt, err := w.offlineTarget(ctx, s, m)

if err != nil {

return stageProgress{}, err

}

+ if tgt.DataBucket != m.StageBucket || tgt.DataBucketRegion != m.StageRegion {

+ return stageProgress{}, fmt.Errorf("the target now publishes %s (%s), not the staging bucket %s (%s) the stage fixed", tgt.DataBucket, tgt.DataBucketRegion, m.StageBucket, m.StageRegion)

+ }

bucket, loc, _, err := cfg.tombstoneAt(host)

if err != nil {

return stageProgress{}, err

}

// The activator promotes a host's moves/<host>/ in its OWN bucket over

// its replica on every restore (promoteStagedReplica: staging wins). A

- // target that publishes this box's own bucket would have the serving

- // instance restore from a copy behind its replica, so the round refuses.

- if tgt.DataBucket == bucket {

- return stageProgress{}, fmt.Errorf("the target publishes this box's own bucket %s; a staged copy there would replace the serving replica on its next wake", bucket)

+ // staging bucket that is this box's own would have the serving

+ // instance restore from a copy behind its replica, so the round refuses

+ // (the stage door refuses it first).

+ if m.StageBucket == bucket {

+ return stageProgress{}, fmt.Errorf("the staging bucket is this box's own bucket %s; a staged copy there would replace the serving replica on its next wake", bucket)

}

- mirror, err := w.offlineMirror(ctx, tgt.DataBucketRegion)

+ mirror, err := w.offlineMirror(ctx, m.StageRegion)

if err != nil {

return stageProgress{}, err

}

+ gated := &stageDeleteGate{MirrorStore: mirror, check: func(ctx context.Context) error {

+ cur, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host)

+ if err != nil {

+ return fmt.Errorf("read the record before the delete phase: %w", err)

+ }

+ if cur.Move == nil || cur.Move.ID != m.ID || cur.Move.State != provision.OfflineStaging {

+ return errors.New("the move left staging during the round; its deletes are not made")

+ }

+ return nil

+ }}

srcPrefix := loc + "/"

dstPrefix := activator.MovesPrefix(host)

receipts := lease.Receipts{Bucket: bucket, Key: activator.MoveReceiptsKey(host)}

- round, err := lease.MirrorPrefix(ctx, mirror, bucket, srcPrefix, tgt.DataBucket, dstPrefix, receipts)

+ round, err := lease.MirrorPrefix(ctx, gated, bucket, srcPrefix, m.StageBucket, dstPrefix, receipts)

if err != nil {

return stageProgress{}, err

}

@@ -145,7 +179,7 @@ func (w *movesWorker) offlineStageRound(ctx context.Context, s release.Store, re

if err != nil {

return stageProgress{}, fmt.Errorf("list the replica: %w", err)

}

- staged, err := mirror.List(ctx, tgt.DataBucket, dstPrefix)

+ staged, err := mirror.List(ctx, m.StageBucket, dstPrefix)

if err != nil {

return stageProgress{}, fmt.Errorf("list the staged copy: %w", err)

}

@@ -165,6 +199,52 @@ func (w *movesWorker) offlineStageRound(ctx context.Context, s release.Store, re

return p, nil

}

+// stageDeleteGate is a stage round's mirror with one check before the

+// delete phase: lease.MirrorPrefix deletes only after every copy has

+// landed, so the first Delete is that phase's start, and check runs there

+// once for all of the round's deletes. Its error fails every delete, so

+// the round deletes nothing.

+type stageDeleteGate struct {

+ lease.MirrorStore

+ check func(ctx context.Context) error

+ once sync.Once

+ err error

+}

+

+func (g *stageDeleteGate) Delete(ctx context.Context, bucket, key string) error {

+ g.once.Do(func() { g.err = g.check(ctx) })

+ if g.err != nil {

+ return g.err

+ }

+ return g.MirrorStore.Delete(ctx, bucket, key)

+}

+

+// offlineStagerStopped is the pinned stager's stop acknowledgement, written

+// in stopping and every later state up to landed, on every pass and every

+// reassert tick, always under this box's pass lock (Astra round 2, finding

+// 1). The record it was handed has left staging, which never comes back,

+// so no stage round starts from here on (offlineStageStep reads the record

+// again before one), and the round that holds the pass lock has ended

+// before this runs. The seal door waits for it, fresh. The progress of the

+// last round is kept.

+func (w *movesWorker) offlineStagerStopped(ctx context.Context, s release.Store, rec provision.Record) error {

+ m := rec.Move

+ if m.StagedAt == nil || m.Stager == "" || m.Stager != w.cfg.Box {

+ return nil

+ }

+ prev, err := w.ownStagerReport(ctx, s, rec)

+ if err != nil {

+ return fmt.Errorf("offline move %s: read the stager report: %w", m.ID, err)

+ }

+ r := moves.OfflineReport{Op: m.ID, Host: rec.Host, Box: w.cfg.Box, Role: moves.OfflineRoleStager, Phase: moves.OfflinePhaseStopped,

+ At: w.cfg.now().UTC(), StageRequest: m.StagedAt.UTC()}

+ if prev != nil {

+ stageProgress{replicaObjects: prev.ReplicaObjects, replicaBytes: prev.ReplicaBytes, stagedObjects: prev.StagedObjects,

+ stagedBytes: prev.StagedBytes, roundObjects: prev.RoundObjects, roundBytes: prev.RoundBytes}.fill(&r)

+ }

+ return writeOfflineReport(ctx, s, r)

+}

+

// ownStagerReport is this box's stager report of the move, nil when it has

// written none. One that does not read as a report is an error.

func (w *movesWorker) ownStagerReport(ctx context.Context, s release.Store, rec provision.Record) (*moves.OfflineReport, error) {

@@ -212,11 +292,18 @@ func (w *movesWorker) offlineUnstage(ctx context.Context, s release.Store, rec p

if own != nil && own.Phase == moves.OfflinePhaseReverted {

return nil

}

- tgt, err := w.offlineTarget(ctx, s, m)

- if err != nil {

- return err

+ // The destination the stage fixed on the record, never what the

+ // target publishes now (Astra round 2, finding 2): reverted is written

+ // only after that prefix is deleted. A stored bucket that is this box's

+ // own was refused at the stage, so it is never deleted from here, and

+ // nothing is acknowledged.

+ if m.StageBucket == "" {

+ return fmt.Errorf("offline move %s: the record names no staging bucket to clean", m.ID)

+ }

+ if m.StageBucket == cfg.OwnBucket {

+ return fmt.Errorf("offline move %s: the stored staging bucket %s is this box's own; nothing is deleted or acknowledged", m.ID, m.StageBucket)

}

- mirror, err := w.offlineMirror(ctx, tgt.DataBucketRegion)

+ mirror, err := w.offlineMirror(ctx, m.StageRegion)

if err != nil {

return err

}

@@ -224,12 +311,10 @@ func (w *movesWorker) offlineUnstage(ctx context.Context, s release.Store, rec p

if prefix == "moves/" || !strings.HasPrefix(prefix, "moves/") {

return fmt.Errorf("offline move %s: no staging prefix for %s", m.ID, host)

}

- if tgt.DataBucket != cfg.OwnBucket {

- if err := lease.DeletePrefix(ctx, mirror, tgt.DataBucket, prefix); err != nil {

- return fmt.Errorf("offline move %s: delete %s/%s: %w", m.ID, tgt.DataBucket, prefix, err)

- }

- log.Printf("moves: %s: offline move %s: the staged copy in %s/%s is deleted", host, m.ID, tgt.DataBucket, prefix)

+ if err := lease.DeletePrefix(ctx, mirror, m.StageBucket, prefix); err != nil {

+ return fmt.Errorf("offline move %s: delete %s/%s: %w", m.ID, m.StageBucket, prefix, err)

}

+ log.Printf("moves: %s: offline move %s: the staged copy in %s/%s is deleted", host, m.ID, m.StageBucket, prefix)

clearOfflineReceipts(ctx, mirror, cfg.OwnBucket, host)

r := moves.OfflineReport{Op: m.ID, Host: host, Box: cfg.Box, Role: moves.OfflineRoleStager, Phase: moves.OfflinePhaseReverted, At: cfg.now().UTC()}

return writeOfflineReport(ctx, s, r)

diff --git a/internal/host/movesoffline_stage_test.go b/internal/host/movesoffline_stage_test.go

index 559cf448b..6cd1f98ba 100644

--- a/internal/host/movesoffline_stage_test.go

+++ b/internal/host/movesoffline_stage_test.go

@@ -54,7 +54,7 @@ func (sw *sealWorld) stage() { sw.stageBy("ie-a") }

// stageBy stages with the stager the door pinned.

func (sw *sealWorld) stageBy(stager string) {

- sw.move(provision.OfflineStage(offID, "", "uk", "op@example.com", stager, sw.clock()))

+ sw.move(provision.OfflineStage(offID, "", "uk", "op@example.com", provision.StageTarget{Stager: stager, Bucket: sealUK, Region: sealUKRegion}, sw.clock()))

}

// restage is a second carlos move offline stage: one more round.

@@ -319,18 +319,17 @@ func TestOfflineAbortFromStagingRemovesTheStagedCopy(t *testing.T) {

}

-// A target that publishes the source's own bucket would have the serving

-// instance restore from the staged copy (the activator's staging wins), so

-// the round refuses and says why in its report.

+// A staging bucket that is the source's own would have the serving instance

+// restore from the staged copy (the activator's staging wins). The stage

+// door refuses it (MV-O191); a record that names it anyway (written by

+// hand) gets a refusing round that writes nothing, and a cleanup that

+// deletes and acknowledges nothing.

func TestOfflineStageRefusesTheSourcesOwnBucket(t *testing.T) {

sw := newSealWorld(t)

ie := sw.ie

sw.instance(ie)

- var last []byte

- if err := edgesFileWrite(context.Background(), sw.s, sw.uk.reg, edgesFile{Box: "uk-box", Placement: "uk", DataBucket: sealIE, DataBucketRegion: sealIERegion}, "", &last, false); err != nil {

- t.Fatal(err)

- }

- sw.stage()

+ sw.publishTarget(sealIE, sealIERegion)

+ sw.move(provision.OfflineStage(offID, "", "uk", "op@example.com", provision.StageTarget{Stager: "ie-a", Bucket: sealIE, Region: sealIERegion}, sw.clock()))

if err := sw.stagePass(ie); err == nil {

t.Fatal("staged into the source's own bucket")

}

@@ -341,6 +340,13 @@ func TestOfflineStageRefusesTheSourcesOwnBucket(t *testing.T) {

if r.Phase != moves.OfflinePhaseFailed || !strings.Contains(r.StageError, "own bucket") {

t.Fatalf("stager report %+v", r)

}

+ sw.abort()

+ if err := sw.stagePass(ie); err == nil {

+ t.Fatal("an abort acknowledged a cleanup of the source's own bucket")

+ }

+ if r := sw.report(moves.OfflineRoleStager, "ie-a"); r.Phase == moves.OfflinePhaseReverted {

+ t.Fatal("reverted written for a staging bucket that was never cleaned")

+ }

}

// Astra round 1, finding 1: a round that copied and then never wrote its

@@ -415,3 +421,154 @@ func TestOfflineAbortWaitsForThePinnedStager(t *testing.T) {

t.Fatalf("stager report %+v", r)

}

+

+// ---- Astra round 2 ----

+

+// publishTarget republishes the target box's edges file with bucket in

+// region: a target replaced or reconfigured after the stage.

+func (sw *sealWorld) publishTarget(bucket, region string) {

+ sw.t.Helper()

+ var last []byte

+ if err := edgesFileWrite(context.Background(), sw.s, sw.uk.reg, edgesFile{Box: "uk-box", Placement: "uk", DataBucket: bucket, DataBucketRegion: region}, "", &last, false); err != nil {

+ sw.t.Fatal(err)

+ }

+}

+

+// Finding 1: once the move has left staging, the stager never starts a

+// round, even on a pass that read the record while it was still staging.

+func TestOfflineStageNoRoundAfterTheMoveLeftStaging(t *testing.T) {

+ sw := newSealWorld(t)

+ ie := sw.ie

+ sw.instance(ie)

+ sw.stage()

+ sw.mustStagePass(ie)

+ sw.restage()

+ stale := sw.record() // a pass's listing, read while still staging

+ sw.move(provision.OfflineStartStaged(offID, "op@example.com", sw.clock()))

+ sw.fake.Put(sealIE, sealPrefix+"/"+offHost+"/scratch/new.bin", []byte("new"))

+ spy := sw.spyCopies()

+ if err := ie.w.offlineRound(context.Background(), sw.s, stale); err != nil {

+ t.Fatal(err)

+ }

+ if n := len(spy.take()); n != 0 {

+ t.Fatalf("a round made %d copies after the move left staging", n)

+ }

+ if r := sw.report(moves.OfflineRoleStager, "ie-a"); r.StageRequest.Equal(*stale.Move.StagedAt) {

+ t.Fatal("the stager answered a request after the move left staging")

+ }

+}

+

+// Finding 1: a round the start overtakes deletes nothing. The delete phase

+// reads the record again first, and a move that has left staging stops the

+// round there: an object the seal's checkpoint added must never be removed

+// by a stage round.

+func TestOfflineStageRoundDeletesNothingOnceTheMoveLeftStaging(t *testing.T) {

+ sw := newSealWorld(t)

+ ie := sw.ie

+ sw.instance(ie)

+ sw.stage()

+ sw.mustStagePass(ie)

+ // An object only the staging prefix has (the round would delete it),

+ // and a new replica object (the round copies it, and the start lands

+ // during that copy).

+ extra := activator.MovesPrefix(offHost) + "0000/added-by-the-seal.ltx"

+ sw.fake.Put(sealUK, extra, []byte("the seal's"))

+ sw.fake.Put(sealIE, sealPrefix+"/"+offHost+"/scratch/new.bin", []byte("new"))

+ sw.restage()

+ var once sync.Once

+ sw.fake.FailOn = func(op, key string) error {

+ if op == "copy" {

+ once.Do(func() {

+ go func() { // the operator's start, while the copy is in flight

+ _, _ = provision.SetRecordMove(context.Background(), sw.s, sw.acct, "app", offHost,

+ provision.OfflineStartStaged(offID, "op@example.com", time.Now()))

+ }()

+ time.Sleep(200 * time.Millisecond)

+ })

+ }

+ return nil

+ }

+ t.Cleanup(func() { sw.fake.FailOn = nil })

+ _ = ie.w.offlineRound(context.Background(), sw.s, sw.record())

+ if sw.record().Move.State != provision.OfflineStopping {

+ t.Fatal("fixture: the start did not land during the round")

+ }

+ if _, ok := sw.fake.Current(sealUK, extra); !ok {

+ t.Fatal("a stage round deleted from the staging prefix after the move left staging")

+ }

+}

+

+// Finding 1: in stopping and on, the pinned stager says, on every pass and

+// under its pass lock, that no round is in flight and none will start.

+func TestOfflineStagerAcknowledgesItHasStopped(t *testing.T) {

+ sw := newSealWorld(t)

+ ie := sw.ie

+ sw.instance(ie)

+ sw.stage()

+ sw.mustStagePass(ie)

+ sw.move(provision.OfflineStartStaged(offID, "op@example.com", sw.clock()))

+ ie.tick()

+ r := sw.report(moves.OfflineRoleStager, "ie-a")

+ if r.Phase != moves.OfflinePhaseStopped || !r.At.Equal(sw.clock()) || r.StagedObjects == 0 {

+ t.Fatalf("stager report %+v, want stopped now with its progress kept", r)

+ }

+}

+

+// Finding 2: cleanup deletes from the destination the stage fixed on the

+// record, whatever the target publishes now, and says reverted only then.

+func TestOfflineAbortCleansTheStoredDestination(t *testing.T) {

+ for _, c := range []struct {

+ name string

+ bucket, region string

+ }{

+ {"a replaced target", "data-uk-new", sealUKRegion},

+ {"a target that now publishes the source's own bucket", sealIE, sealIERegion},

+ } {

+ t.Run(c.name, func(t *testing.T) {

+ sw := newSealWorld(t)

+ sw.fake.AddBucket("data-uk-new")

+ ie := sw.ie

+ sw.instance(ie)

+ sw.stage()

+ sw.mustStagePass(ie)

+ if len(sw.stagedRel()) == 0 {

+ t.Fatal("fixture: nothing staged")

+ }

+ sw.publishTarget(c.bucket, c.region)

+ sw.abort()

+ sw.mustStagePass(ie)

+ if n := len(sw.stagedRel()); n != 0 {

+ t.Fatalf("%d objects left in the stored staging bucket %s", n, sealUK)

+ }

+ if r := sw.report(moves.OfflineRoleStager, "ie-a"); r.Phase != moves.OfflinePhaseReverted {

+ t.Fatalf("stager report %+v", r)

+ }

+ if got := sw.replicaRel(); len(got) == 0 {

+ t.Fatal("the cleanup touched the source's replica")

+ }

+ })

+ }

+}

+

+// Finding 2: a stage round never follows a target that now publishes other

+// coordinates than the stored ones. It refuses and writes nowhere.

+func TestOfflineStageRoundRefusesChangedCoordinates(t *testing.T) {

+ sw := newSealWorld(t)

+ sw.fake.AddBucket("data-uk-new")

+ ie := sw.ie

+ sw.instance(ie)

+ sw.stage()

+ sw.publishTarget("data-uk-new", sealUKRegion)

+ if err := sw.stagePass(ie); err == nil {

+ t.Fatal("staged with changed coordinates")

+ }

+ if n := len(sw.list("data-uk-new", "moves/")); n != 0 {

+ t.Fatalf("%d objects staged into the new bucket", n)

+ }

+ if n := len(sw.stagedRel()); n != 0 {

+ t.Fatalf("%d objects staged at all", n)

+ }

+ if r := sw.report(moves.OfflineRoleStager, "ie-a"); r.Phase != moves.OfflinePhaseFailed {

+ t.Fatalf("stager report %+v", r)

+ }

+}

diff --git a/internal/provision/offline_move.go b/internal/provision/offline_move.go

index 1db2bcf66..8d87fb943 100644

--- a/internal/provision/offline_move.go

+++ b/internal/provision/offline_move.go

@@ -129,20 +129,25 @@ func OfflineStart(id, from, to, by string, now time.Time) func(*Record) error {

return offlineBegin(id, from, to, by, now, OfflineStopping)

}

+// StageTarget is what a stage fixes for the whole move: the one box that

+// copies (Stager) and where the copy goes (Bucket, the target's data

+// bucket, in Region).

+type StageTarget struct{ Stager, Bucket, Region string }

+

// OfflineStage is OfflineStart into offline-staging: the same refusals and

-// the same stamps, plus staged_at and the stager, the one box that copies

-// for this move (a stage that names none is malformed). The instance keeps

+// the same stamps, plus staged_at, the stager and the staging destination

+// (a stage that names any of them empty is malformed). The instance keeps

// serving.

-func OfflineStage(id, from, to, by, stager string, now time.Time) func(*Record) error {

+func OfflineStage(id, from, to, by string, st StageTarget, now time.Time) func(*Record) error {

begin := offlineBegin(id, from, to, by, now, OfflineStaging)

return func(rec *Record) error {

- if stager == "" {

+ if st.Stager == "" || st.Bucket == "" {

return fmt.Errorf("%w: %s", ErrMoveConflict, refusedMalformedStart)

}

if err := begin(rec); err != nil {

return err

}

- rec.Move.Stager = stager

+ rec.Move.Stager, rec.Move.StageBucket, rec.Move.StageRegion = st.Stager, st.Bucket, st.Region

return nil

}

diff --git a/internal/provision/offline_stage_test.go b/internal/provision/offline_stage_test.go

index be44df310..9da62fd55 100644

--- a/internal/provision/offline_stage_test.go

+++ b/internal/provision/offline_stage_test.go

@@ -21,20 +21,20 @@ func TestOfflineStageThenStartThenTheWholeMove(t *testing.T) {

s := release.NewDirStore(t.TempDir())

rec := offlineRecord(t, s)

- got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", "box-a", now))

+ got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now))

if err != nil {

t.Fatal(err)

}

m := got.Move

if !m.Offline() || m.State != OfflineStaging || m.ID != offID || m.To != "uk" || got.Region != "" ||

- m.StagedAt == nil || !m.StagedAt.Equal(now) || !m.RequestedAt.Equal(now) || m.Stager != "box-a" {

+ m.StagedAt == nil || !m.StagedAt.Equal(now) || !m.RequestedAt.Equal(now) || m.Stager != "box-a" || m.StageBucket != "data-uk" || m.StageRegion != "eu-west-2" {

t.Fatalf("staged record: %+v region %q", m, got.Region)

}

// Staging is a move in flight: no second move starts beside it.

if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStart(otherOff, "", "uk", "ops@example.com", now)); !errors.Is(err, ErrMoveConflict) {

t.Fatalf("a start beside a staged move: %v", err)

}

- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(otherOff, "", "us", "ops@example.com", "box-a", now)); !errors.Is(err, ErrMoveConflict) {

+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(otherOff, "", "us", "ops@example.com", stageAt, now)); !errors.Is(err, ErrMoveConflict) {

t.Fatalf("a stage beside a staged move: %v", err)

}

@@ -74,7 +74,7 @@ func TestOfflineStartStagedRefusals(t *testing.T) {

}

s := release.NewDirStore(t.TempDir())

rec := offlineRecord(t, s)

- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", "box-a", now)); err != nil {

+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now)); err != nil {

t.Fatal(err)

}

if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(otherOff, "ops@example.com", now)); !errors.Is(err, ErrMoveConflict) {

@@ -91,7 +91,7 @@ func TestOfflineStartStagedRefusals(t *testing.T) {

// No other step leaves staging: seal through finish all refuse.

s = release.NewDirStore(t.TempDir())

rec = offlineRecord(t, s)

- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", "box-a", now)); err != nil {

+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now)); err != nil {

t.Fatal(err)

}

before := offlineGet(t, s, rec)

@@ -112,7 +112,7 @@ func TestOfflineStartStagedIsAlwaysAfterTheStage(t *testing.T) {

now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)

s := release.NewDirStore(t.TempDir())

rec := offlineRecord(t, s)

- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", "box-a", now)); err != nil {

+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now)); err != nil {

t.Fatal(err)

}

got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStartStaged(offID, "ops@example.com", now.Add(-time.Second)))

@@ -132,7 +132,7 @@ func TestOfflineAbortFromStaging(t *testing.T) {

now := time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC)

s := release.NewDirStore(t.TempDir())

rec := offlineRecord(t, s)

- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", "box-a", now)); err != nil {

+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now)); err != nil {

t.Fatal(err)

}

got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineAbort(offID))

@@ -147,7 +147,7 @@ func TestOfflineAbortFromStaging(t *testing.T) {

s = release.NewDirStore(t.TempDir())

rec = offlineRecord(t, s)

for _, mut := range []func(*Record) error{

- OfflineStage(offID, "", "uk", "ops@example.com", "box-a", now),

+ OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now),

OfflineStartStaged(offID, "ops@example.com", now.Add(20*time.Minute)),

OfflineAbort(offID),

} {

@@ -164,7 +164,7 @@ func TestOfflineAbortFromStaging(t *testing.T) {

// staged_at is additive like the other offline fields: a staged journal

// round-trips, and every earlier shape still does (TestOfflineFieldsAreAdditiveOnTheWire).

func TestOfflineStagedAtIsAdditiveOnTheWire(t *testing.T) {

- const staged = `{"schema":1,"host":"a.go.example.com","slug":"s","app":"app","account":"bab","region":"ie","channel":"stable","created_at":"2026-09-01T08:00:00Z","move":{"id":"mv_abcdefghijklmnopqrstuvwxyz","from":"ie","to":"uk","state":"offline-staging","requested_by":"ops@example.com","requested_at":"2026-10-21T09:00:00Z","updated_at":"2026-10-21T09:00:00Z","kind":"offline","staged_at":"2026-10-21T09:00:00Z","stager":"box-a"}}`

+ const staged = `{"schema":1,"host":"a.go.example.com","slug":"s","app":"app","account":"bab","region":"ie","channel":"stable","created_at":"2026-09-01T08:00:00Z","move":{"id":"mv_abcdefghijklmnopqrstuvwxyz","from":"ie","to":"uk","state":"offline-staging","requested_by":"ops@example.com","requested_at":"2026-10-21T09:00:00Z","updated_at":"2026-10-21T09:00:00Z","kind":"offline","staged_at":"2026-10-21T09:00:00Z","stager":"box-a","stage_bucket":"data-uk","stage_region":"eu-west-2"}}`

rec, err := DecodeRecord([]byte(staged))

if err != nil {

t.Fatal(err)

@@ -188,7 +188,7 @@ func TestOfflineRestage(t *testing.T) {

if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineRestage(offID, now)); !errors.Is(err, ErrMoveConflict) {

t.Fatalf("restage at rest: %v", err)

}

- if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", "box-a", now)); err != nil {

+ if _, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", stageAt, now)); err != nil {

t.Fatal(err)

}

got, err := SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, OfflineRestage(offID, now.Add(5*time.Minute)))

@@ -213,14 +213,21 @@ func TestOfflineRestage(t *testing.T) {

}

-// The stager is pinned at the stage: a stage that names none is malformed.

+// stageAt is the stager and staging destination the tests' stage door names.

+var stageAt = StageTarget{Stager: "box-a", Bucket: "data-uk", Region: "eu-west-2"}

+

+// The stager is pinned at the stage: a stage that names none is malformed,

+// and so is one with no staging bucket.

func TestOfflineStageNamesItsStager(t *testing.T) {

s := release.NewDirStore(t.TempDir())

rec := offlineRecord(t, s)

before := offlineGet(t, s, rec)

- if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", "", time.Now())); !errors.Is(err, ErrMoveConflict) {

+ if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", StageTarget{Bucket: "data-uk"}, time.Now())); !errors.Is(err, ErrMoveConflict) {

t.Fatalf("a stage with no stager: %v", err)

}

+ if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, OfflineStage(offID, "", "uk", "ops@example.com", StageTarget{Stager: "box-a"}, time.Now())); !errors.Is(err, ErrMoveConflict) {

+ t.Fatalf("a stage with no staging bucket: %v", err)

+ }

if after := offlineGet(t, s, rec); !reflect.DeepEqual(after, before) {

t.Fatal("a refused stage changed the record")

}

diff --git a/internal/provision/provision.go b/internal/provision/provision.go

index 844cbed4c..a5657b0cb 100644

--- a/internal/provision/provision.go

+++ b/internal/provision/provision.go

@@ -1078,6 +1078,14 @@ type MoveState struct {

// of a staged move finishes only once this box has removed the staged

// copy. Additive and omitempty.

Stager string `json:"stager,omitempty"`

+ // StageBucket and StageRegion are the staged copy's destination, the

+ // target's data bucket and its region as the stage door read them,

+ // fixed before any copy (Astra round 2, finding 2). Every stage round,

+ // the seal's hand-over and an abort's cleanup use them; a target that

+ // later publishes other coordinates is refused, never followed.

+ // Additive and omitempty.

+ StageBucket string `json:"stage_bucket,omitempty"`

+ StageRegion string `json:"stage_region,omitempty"`

}

// MoveInputs (what the target was checked against, §7.3 T3) is NOT

carlos / platform Public

moves offline: Astra round 2, the stager in the seal barrier, a fixed staging destination