amadan

carlos / platform Public

simplify: the residence-move surface after four cleanup angles (#331/#346)

Reuse: one CAS scaffold (updateRecord) under SetRecordRegion and
SetRecordMove, which now returns the record it wrote so the console
renders the view without a third Get; readJSON at the move door (the
console's standard cap/415/413 answers); errors.Is over string-matching
for ErrMoveConflict; lease.DeletePrefix under both staging purges; the
SDK's ListObjectsV2 paginator.

Simplification: moveTarget and the console's dead data_bucket fields
gone — the host worker decodes the writer's own edgesFile; the dead
identity method on s3Mirror; armedCodes renders every refusal list;
edgesFilePass takes the box's identity half of the file as a struct
instead of six positional strings (28 test call sites rewritten).

Efficiency: mirror clients memoized per target region on a movesWorker
(and built lazily — the not-yet-quiesced cut branch never pays a
credential-chain resolution); s3Mirror memoizes bucket→client so only
the first touch pays the wrong-region probe; a landing restore drops
the staged keys promote already listed instead of listing a third time.

Altitude: the worker's three package globals fold into movesWorker
(trouble entries now clear on a clean round); lock.json's exclusion is
declared as lease policy (leaseLockName).

Skipped, deliberately: hoisting ListAllRecords out of the moves pass —
the reconcile pass does not expose its walk, and sharing it is a
reconcile refactor, not a moves cleanup (noted for #363's neighbour);
applying homeRegionOut to the instance GET's region field — a wire
change to a shipped answer titogo already reads as "" = unconstrained.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PEypt7de2X1G84aQ9rNkCf
Paul Campbell pushed by paul@keymail.dev 79bb4f0235b1d64c70a68739a98fdb6a97732c68 parent 48a0178
12 files changed, +270 −217
  • internal/activator/backend_exec.go +13 −17
  • internal/console/apiinstancemove.go +20 −19
  • internal/console/apiinstancemove_test.go +2 −2
  • internal/console/apiinstances.go +2 −9
  • internal/console/instancesplacement.go +6 −8
  • internal/host/edgesfile.go +16 −7
  • internal/host/edgesfile_test.go +28 −28
  • internal/host/movespass.go +74 −57
  • internal/host/movespass_test.go +19 −12
  • internal/lease/mirror.go +57 −21
  • internal/provision/provision.go +31 −36
  • internal/provision/provision_test.go +2 −1
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 995870cbf..3e0d41be6 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -243,7 +243,7 @@ func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, err
// 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
+ var promoted map[string]string
if b.Mirror != nil {
var err error
if promoted, err = b.promoteStagedReplica(ctx, host); err != nil {
@@ -299,8 +299,8 @@ func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, err
// 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)
+ if len(promoted) > 0 {
+ b.dropStagedReplica(ctx, promoted)
}
} else {
// Never replicated: a brand-new instance boots fresh and creates its
@@ -1059,7 +1059,7 @@ func PurgeLocalDBArtifacts(dbPath string, full bool) {
// 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) {
+func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (map[string]string, 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
@@ -1074,27 +1074,23 @@ func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (bo
// keys answer for app.example.co.
staged, err := b.Mirror.List(ctx, b.Cfg.Bucket, MovesPrefix(host))
if err != nil {
- return false, err
+ return nil, err
}
if len(staged) == 0 {
- return false, nil
+ return nil, 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 nil, err
}
- return true, nil
+ return staged, 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
- }
+// dropStagedReplica deletes the staged keys promote just returned — no
+// third listing of a prefix this restore has already walked twice.
+// Best-effort: a leftover staging copy is storage, not a correctness
+// problem.
+func (b *ExecBackend) dropStagedReplica(ctx context.Context, staged map[string]string) {
for k := range staged {
_ = b.Mirror.Delete(ctx, b.Cfg.Bucket, k)
}
diff --git a/internal/console/apiinstancemove.go b/internal/console/apiinstancemove.go
index d156cc921..23e674b7c 100644
--- a/internal/console/apiinstancemove.go
+++ b/internal/console/apiinstancemove.go
@@ -1,7 +1,6 @@
package console
import (
- "encoding/json"
"errors"
"log"
"net/http"
@@ -97,6 +96,20 @@ func (s *Server) moveViewOf(rec *provision.Record) moveView {
}
}
+// armedCodes renders the armed half of an armedPlacements answer as the
+// sorted comma list every refusal names — one spelling for the create door
+// and the move door alike.
+func armedCodes(set map[string]bool) string {
+ codes := make([]string, 0, len(set))
+ for c, armed := range set {
+ if armed {
+ codes = append(codes, c)
+ }
+ }
+ slices.Sort(codes)
+ return strings.Join(codes, ", ")
+}
+
// homeRegionIn/Out translate between the home region's NAME (Config's
// CONSOLE_HOME_REGION, what pickers and callers speak — issue #346) and the
// record's "" (the unconstrained value the fleet's reconcile has always
@@ -120,7 +133,7 @@ func (s *Server) homeRegionOut(region string) string {
// flat move view + 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)
+ rec, 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
@@ -131,11 +144,6 @@ func (s *Server) moveTransition(w http.ResponseWriter, r *http.Request, acct *Ac
return
}
s.bumpSerial(ctx)
- rec, err := provision.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app, host)
- if err != nil {
- writeJSON(w, okStatus, moveView{State: "none"})
- return
- }
writeJSON(w, okStatus, s.moveViewOf(rec))
}
@@ -157,12 +165,12 @@ func (s *Server) handleAPIInstanceMoveRequest(w http.ResponseWriter, r *http.Req
var req struct {
Region string `json:"region"`
}
- if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
- jsonError(w, http.StatusBadRequest, "invalid JSON body")
+ if !readJSON(w, r, &req) {
return
}
- to := s.homeRegionIn(strings.TrimSpace(req.Region))
- if strings.TrimSpace(req.Region) == "" || !provision.ValidRegion(to) {
+ region := strings.TrimSpace(req.Region)
+ to := s.homeRegionIn(region)
+ if region == "" || !provision.ValidRegion(to) {
jsonError(w, http.StatusBadRequest, "invalid region")
return
}
@@ -182,14 +190,7 @@ func (s *Server) handleAPIInstanceMoveRequest(w http.ResponseWriter, r *http.Req
return
}
if !set[to] {
- codes := make([]string, 0, len(set))
- for c, armed := range set {
- if armed {
- codes = append(codes, c)
- }
- }
- slices.Sort(codes)
- jsonError(w, http.StatusBadRequest, "region "+to+" is not armed in this deployment — armed: "+strings.Join(codes, ", "))
+ jsonError(w, http.StatusBadRequest, "region "+to+" is not armed in this deployment — armed: "+armedCodes(set))
return
}
s.moveTransition(w, r, acct, app, host, provision.MoveRequest(to, time.Now()), http.StatusAccepted)
diff --git a/internal/console/apiinstancemove_test.go b/internal/console/apiinstancemove_test.go
index 247322ed1..6b4a8a136 100644
--- a/internal/console/apiinstancemove_test.go
+++ b/internal/console/apiinstancemove_test.go
@@ -105,7 +105,7 @@ func TestAPIInstanceMoveVerbs(t *testing.T) {
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 {
+ 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, "")
@@ -115,7 +115,7 @@ func TestAPIInstanceMoveVerbs(t *testing.T) {
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 {
+ 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 {
diff --git a/internal/console/apiinstances.go b/internal/console/apiinstances.go
index 561429364..b18c33f07 100644
--- a/internal/console/apiinstances.go
+++ b/internal/console/apiinstances.go
@@ -616,16 +616,9 @@ func (s *Server) handleAPIInstanceCreate(w http.ResponseWriter, r *http.Request)
// dormant until some box publishes a placement code, so a console rolled
// ahead of its fleet changes nothing.
if placed == "" && region != "" {
- if set, armed := armedPlacements(ctx, s.cfg.Releases); armed && !set[region] {
- codes := make([]string, 0, len(set))
- for c, armed := range set {
- if armed {
- codes = append(codes, c)
- }
- }
- slices.Sort(codes)
+ if set, published := armedPlacements(ctx, s.cfg.Releases); published && !set[region] {
jsonError(w, http.StatusBadRequest,
- "region "+region+" is not armed in this deployment — armed: "+strings.Join(codes, ", "))
+ "region "+region+" is not armed in this deployment — armed: "+armedCodes(set))
return
}
}
diff --git a/internal/console/instancesplacement.go b/internal/console/instancesplacement.go
index 26865c727..41969cf15 100644
--- a/internal/console/instancesplacement.go
+++ b/internal/console/instancesplacement.go
@@ -33,14 +33,12 @@ 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"`
- DataBucket string `json:"data_bucket"`
- DataBucketRegion string `json:"data_bucket_region"`
- 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"`
+ Hosts map[string]json.RawMessage `json:"hosts"`
}
// loadEdgesJoin folds every readable edges file into host → the boxes that
diff --git a/internal/host/edgesfile.go b/internal/host/edgesfile.go
index dd4ec7685..9c4e86bdb 100644
--- a/internal/host/edgesfile.go
+++ b/internal/host/edgesfile.go
@@ -131,7 +131,12 @@ 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, dataBucket, dataBucketRegion string, last *[]byte) error {
+// edgesFilePass publishes one file. ident is this box's IDENTITY half of
+// the file — Box, Region, Placement, DataBucket, DataBucketRegion — built
+// once by the loop; eip is passed per pass because it is re-probed every
+// tick and held across blanks (below). Hosts are derived here.
+func edgesFilePass(ctx context.Context, s release.Store, reg *registry.Registry, ident edgesFile, eip string, last *[]byte) error {
+ box := ident.Box
routes, err := reg.List()
if err != nil {
return fmt.Errorf("edges: list: %w", err)
@@ -265,8 +270,8 @@ func edgesFilePass(ctx context.Context, s release.Store, reg *registry.Registry,
}
}
- f := edgesFile{V: 1, Box: box, EIP: eip, Region: region, Placement: placement,
- DataBucket: dataBucket, DataBucketRegion: dataBucketRegion, Hosts: hosts}
+ f := ident
+ f.V, f.EIP, f.Hosts, f.Generation = 1, eip, hosts, ""
gate, err := json.Marshal(f)
if err != nil {
return fmt.Errorf("edges: marshal: %w", err)
@@ -380,9 +385,13 @@ func edgesFileLoop(reg *registry.Registry) {
// The placement code alongside it: CARLOS_REGION, the same env the
// 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"))
+ ident := edgesFile{
+ Box: box,
+ Region: region,
+ 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
@@ -390,7 +399,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, dataBucket, dataBucketRegion, &last); err != nil {
+ if err := edgesFilePass(ctx, store, reg, ident, eip, &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 db859c8e8..96fd900ff 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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "", &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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "", &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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "quiet-harbour", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "quiet-harbour", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "quiet-harbour", Region: "eu-west-1", Placement: "", DataBucket: "", DataBucketRegion: ""}, "203.0.113.10", &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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "", &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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "", &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, edgesFile{Box: "annie", Region: "eu-west-1", Placement: "", DataBucket: "", DataBucketRegion: ""}, "1.2.3.4", &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, edgesFile{Box: "annie", Region: "eu-west-1", Placement: "", DataBucket: "", DataBucketRegion: ""}, "1.2.3.4", &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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: "annie", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "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, edgesFile{Box: box, Region: "eu-west-1", Placement: "", DataBucket: "", DataBucketRegion: ""}, "203.0.113.7", &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, edgesFile{Box: "quiet-harbour", Region: "eu-west-2", Placement: "uk", DataBucket: "", DataBucketRegion: ""}, "203.0.113.10", &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, edgesFile{Box: "quiet-harbour", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "203.0.113.10", &last2); err != nil {
t.Fatal(err)
}
got2, _, err := s2.Get(context.Background(), "control/edges/quiet-harbour.json")
@@ -860,7 +860,7 @@ 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 {
+ if err := edgesFilePass(context.Background(), s, reg, edgesFile{Box: "quiet-harbour", Region: "eu-west-2", Placement: "uk", DataBucket: "carlos-tito-uk-1", DataBucketRegion: "eu-west-2"}, "203.0.113.10", &last); err != nil {
t.Fatal(err)
}
got, _, err := s.Get(context.Background(), "control/edges/quiet-harbour.json")
@@ -880,7 +880,7 @@ func TestEdgesFileRawWireBodyDataBucketFields(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, edgesFile{Box: "quiet-harbour", Region: "", Placement: "", DataBucket: "", DataBucketRegion: ""}, "203.0.113.10", &last2); err != nil {
t.Fatal(err)
}
got2, _, err := s2.Get(context.Background(), "control/edges/quiet-harbour.json")
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index f61190ba9..c3d195093 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -3,6 +3,7 @@ package host
import (
"context"
"encoding/json"
+ "errors"
"fmt"
"log"
"os"
@@ -42,11 +43,37 @@ import (
// 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]
+// movesWorker is the one home for the worker's cross-pass state: the wired
+// config, the memoized per-region mirror clients (built once per target
+// region, not per pass — a client build is a credential-chain resolution),
+// the once-per-host status nudge, and the per-host trouble throttle.
+// RunAgent writes the pointer once; both converge loops read it.
+type movesWorker struct {
+ cfg moveCfg
+ mu sync.Mutex
+ mirrors map[string]lease.MirrorStore // dstRegion → client pair
+ nudged sync.Map // host → true: status→quiesce hop nudged
+ trouble sync.Map // host → time.Time of last logged error
+}
+
+var armedMovesWorker atomic.Pointer[movesWorker]
+
+func (w *movesWorker) mirror(ctx context.Context, dstRegion string) (lease.MirrorStore, error) {
+ w.mu.Lock()
+ m := w.mirrors[dstRegion]
+ w.mu.Unlock()
+ if m != nil {
+ return m, nil
+ }
+ m, err := w.cfg.Mirror(ctx, dstRegion)
+ if err != nil {
+ return nil, err
+ }
+ w.mu.Lock()
+ w.mirrors[dstRegion] = m
+ w.mu.Unlock()
+ return m, nil
+}
// armMovesPass wires the worker with what only RunAgent has in hand: the
// activator's hibernate coordinates (the own-bucket side of every mirror),
@@ -54,6 +81,7 @@ var armedMoveCfg atomic.Pointer[moveCfg]
// 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,
@@ -80,17 +108,17 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry,
},
Now: time.Now,
}
- armedMoveCfg.Store(&cfg)
+ armedMovesWorker.Store(&movesWorker{cfg: cfg, mirrors: map[string]lease.MirrorStore{}})
}
// 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 {
+ w := armedMovesWorker.Load()
+ if w == nil || s == nil {
return
}
- if err := movesPass(ctx, s, *cfg); err != nil {
+ if err := movesPass(ctx, s, w); err != nil {
log.Printf("agent: moves pass: %v", err)
}
}
@@ -120,23 +148,14 @@ type moveCfg struct {
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 {
+func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {
+ cfg := w.cfg
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
+ var targets map[string]edgesFile // lazy: most passes carry no move
for _, rec := range recs {
if rec.Move == nil || rec.Region != cfg.Region {
continue
@@ -159,13 +178,15 @@ func movesPass(ctx context.Context, s release.Store, cfg moveCfg) error {
return nil
}
}
- if err := runMoveRound(ctx, s, cfg, rec, targets); err != nil {
+ if err := runMoveRound(ctx, s, w, 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)
+ w.logTrouble(rec.Host, err)
+ continue
}
+ w.trouble.Delete(rec.Host)
}
return nil
}
@@ -174,14 +195,14 @@ func movesPass(ctx context.Context, s release.Store, cfg moveCfg) error {
// 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) {
+func moveTargets(ctx context.Context, s release.Store) (map[string]edgesFile, error) {
bodies, err := loadEdgesFileBodies(ctx, s)
if err != nil {
return nil, err
}
- out := map[string]moveTarget{}
+ out := map[string]edgesFile{}
for _, body := range bodies {
- var t moveTarget
+ var t edgesFile // the writer's own shape — no parallel decl to drift
if err := json.Unmarshal(body, &t); err != nil || t.V < 1 || t.Placement == "" || t.DataBucket == "" {
continue
}
@@ -193,27 +214,31 @@ func moveTargets(ctx context.Context, s release.Store) (map[string]moveTarget, e
return out, nil
}
-func runMoveRound(ctx context.Context, s release.Store, cfg moveCfg, rec provision.Record, targets map[string]moveTarget) error {
+func runMoveRound(ctx context.Context, s release.Store, w *movesWorker, rec provision.Record, targets map[string]edgesFile) error {
+ cfg := w.cfg
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)
- }
+ // The mirror pair is built lazily and memoized per target region: the
+ // cut's not-yet-quiesced early return must not pay a credential-chain
+ // resolution it never uses.
+ mirror := func() (lease.MirrorStore, error) { return w.mirror(ctx, tgt.DataBucketRegion) }
srcPrefix := activator.HibernatePrefix(cfg.OwnPrefix, rec.Host)
dstPrefix := activator.MovesPrefix(rec.Host)
switch rec.Move.State {
case provision.MovePreparing, provision.MoveReady:
+ m, err := mirror()
+ if err != nil {
+ return err
+ }
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) {
+ if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MovePromoteReady(cfg.Now())); err != nil && !errors.Is(err, provision.ErrMoveConflict) {
return err
}
nudgeSerial(ctx, s)
@@ -225,17 +250,14 @@ func runMoveRound(ctx context.Context, s release.Store, cfg moveCfg, rec provisi
// 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)
+ m, err := mirror()
if err != nil {
return err
}
- for k := range staged {
- if err := m.Delete(ctx, tgt.DataBucket, k); err != nil {
- return err
- }
+ if err := lease.DeletePrefix(ctx, m, tgt.DataBucket, dstPrefix); err != nil {
+ return err
}
- err = provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveAbortComplete())
- if err != nil && !isMoveConflict(err) {
+ if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveAbortComplete()); err != nil && !errors.Is(err, provision.ErrMoveConflict) {
return err
}
nudgeSerial(ctx, s)
@@ -249,7 +271,7 @@ func runMoveRound(ctx context.Context, s release.Store, cfg moveCfg, rec provisi
return fmt.Errorf("set route status: %w", err)
}
if !cfg.Quiesced(rec.Host) {
- if _, done := movingStatusNudged.LoadOrStore(rec.Host, true); !done {
+ if _, done := w.nudged.LoadOrStore(rec.Host, true); !done {
nudgeSerial(ctx, s)
}
return nil // the retire is in flight; the nudged pass re-enters here
@@ -269,17 +291,20 @@ func runMoveRound(ctx context.Context, s release.Store, cfg moveCfg, rec provisi
}
}
// Step 4: the byte-final mirror round.
+ m, err := mirror()
+ if err != nil {
+ return err
+ }
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) {
+ if _, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, provision.MoveComplete()); err != nil && !errors.Is(err, provision.ErrMoveConflict) {
return err
}
- movingStatusNudged.Delete(rec.Host)
+ w.nudged.Delete(rec.Host)
// The flip's nudge is what turns the TARGET's mint from a
// 5-minute wait into a seconds-cadence converge.
nudgeSerial(ctx, s)
@@ -308,21 +333,13 @@ func nudgeSerial(ctx context.Context, s release.Store) {
// per process, then the 5-minute floor stands under a wedged stop.
var movingStatusNudged sync.Map
-// 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) {
+// logTrouble rate-limits per-host move errors to one line a minute; the
+// pass clears a host's entry on its next clean round.
+func (w *movesWorker) logTrouble(host string, err error) {
now := time.Now()
- if last, ok := moveTroubleLast.Load(host); ok && now.Sub(last.(time.Time)) < time.Minute {
+ if last, ok := w.trouble.Load(host); ok && now.Sub(last.(time.Time)) < time.Minute {
return
}
- moveTroubleLast.Store(host, now)
+ w.trouble.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
index 3772d3c9f..4cd60ecd7 100644
--- a/internal/host/movespass_test.go
+++ b/internal/host/movespass_test.go
@@ -86,6 +86,13 @@ func newMoveHarness(t *testing.T) *moveHarness {
return h
}
+// worker wraps the harness cfg the way armMovesPass does in production —
+// a fresh worker per call, so memoized mirrors/nudges never leak between
+// the table's cases.
+func (h *moveHarness) worker() *movesWorker {
+ return &movesWorker{cfg: h.cfg, mirrors: map[string]lease.MirrorStore{}}
+}
+
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",
@@ -94,7 +101,7 @@ func (h *moveHarness) record(t *testing.T, host string, mut ...func(*provision.R
t.Fatal(err)
}
for _, m := range mut {
- if err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, m); err != nil {
+ if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, m); err != nil {
t.Fatal(err)
}
}
@@ -124,7 +131,7 @@ func TestMovesPassPrepareAndPromote(t *testing.T) {
// 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 {
+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
t.Fatal(err)
}
if got := h.read(t, host); got.Move.State != provision.MovePreparing {
@@ -135,7 +142,7 @@ func TestMovesPassPrepareAndPromote(t *testing.T) {
t.Fatalf("staging = %v", staged)
}
// Round 2: nothing to do → ready.
- if err := movesPass(context.Background(), h.s, h.cfg); err != nil {
+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
t.Fatal(err)
}
got := h.read(t, host)
@@ -144,7 +151,7 @@ func TestMovesPassPrepareAndPromote(t *testing.T) {
}
// 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 {
+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
t.Fatal(err)
}
staged, _ = h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
@@ -166,7 +173,7 @@ func TestMovesPassCutSequence(t *testing.T) {
// 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 {
+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
t.Fatal(err)
}
if len(h.statuses) != 1 || h.statuses[0] != host+"=moving" {
@@ -183,7 +190,7 @@ func TestMovesPassCutSequence(t *testing.T) {
// 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 {
+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
t.Fatal(err)
}
ckIdx, cpIdx := -1, -1
@@ -211,7 +218,7 @@ func TestMovesPassCutSequence(t *testing.T) {
}
// 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 {
+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
t.Fatal(err)
}
if len(h.f.ops) != before {
@@ -224,7 +231,7 @@ func TestMovesPassSkipsWithoutTargetCoords(t *testing.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 {
+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
t.Fatal(err)
}
if got := h.read(t, host); got.Move.State != provision.MovePreparing {
@@ -248,7 +255,7 @@ func TestMovesPassRailsAndOwnership(t *testing.T) {
// 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 {
+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
t.Fatal(err)
}
if len(h.f.ops) != 0 || h.read(t, host).Move.State != provision.MovePreparing {
@@ -264,10 +271,10 @@ func TestMovesPassRailsAndOwnership(t *testing.T) {
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 {
+ 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 {
+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
t.Fatal(err)
}
if h.read(t, pooled).Move.State != provision.MovePreparing {
@@ -281,7 +288,7 @@ func TestMovesPassAbortPhase(t *testing.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 {
+ if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
t.Fatal(err)
}
staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
diff --git a/internal/lease/mirror.go b/internal/lease/mirror.go
index 684bdf312..2061a4c4e 100644
--- a/internal/lease/mirror.go
+++ b/internal/lease/mirror.go
@@ -6,6 +6,7 @@ import (
"net/url"
"sort"
"strings"
+ "sync"
awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
)
@@ -73,7 +74,7 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB
// any awake instance — and promoting it would hand the target a
// stale foreign lease. Skipped on both sides: never copied, and
// the destination's own copy (if any) is never deleted.
- if r == "lock.json" {
+ if r == leaseLockName {
continue
}
sk, dk := srcPrefix+r, dstPrefix+r
@@ -87,7 +88,7 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB
}
for dk := range dst {
r := dk[len(dstPrefix):]
- if r == "lock.json" {
+ if r == leaseLockName {
continue
}
if _, ok := src[srcPrefix+r]; !ok {
@@ -100,12 +101,40 @@ func MirrorPrefix(ctx context.Context, m MirrorStore, srcBucket, srcPrefix, dstB
return changed, nil
}
+// leaseLockName is the coordination file this package's own leaser keeps
+// beside every replica. MirrorPrefix's exclusion of it is lease POLICY, not
+// generic mirroring — which is why the mirror lives in this package.
+const leaseLockName = "lock.json"
+
+// DeletePrefix removes every object under prefix (the lease lock included:
+// a deleted staging area holds nothing back). The two callers are abort's
+// box half (errors surface — the abort retries) and the activator's
+// post-restore cleanup (best-effort at the call site).
+func DeletePrefix(ctx context.Context, m MirrorStore, bucket, prefix string) error {
+ objs, err := m.List(ctx, bucket, prefix)
+ if err != nil {
+ return err
+ }
+ for k := range objs {
+ if err := m.Delete(ctx, bucket, k); err != nil {
+ return err
+ }
+ }
+ return 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
+ // byBucket memoizes which client answered for a bucket, so only the
+ // FIRST List of each bucket pays the wrong-region probe — and a later
+ // transient error on the right client is not retried against a region
+ // that can only answer with a redirect.
+ mu sync.Mutex
+ byBucket map[string]*awss3.Client
}
// NewS3Mirror builds the two-region pair with the same credential chain and
@@ -120,18 +149,14 @@ func NewS3Mirror(ctx context.Context, srcRegion, dstRegion string) (MirrorStore,
if err != nil {
return nil, err
}
- return &s3Mirror{src: src, dst: dst}, nil
+ return &s3Mirror{src: src, dst: dst, byBucket: map[string]*awss3.Client{}}, 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,
- })
+ pager := awss3.NewListObjectsV2Paginator(c, &awss3.ListObjectsV2Input{Bucket: &bucket, Prefix: &prefix})
+ for pager.HasMorePages() {
+ resp, err := pager.NextPage(ctx)
if err != nil {
return nil, err
}
@@ -145,22 +170,33 @@ func (m *s3Mirror) list(ctx context.Context, c *awss3.Client, bucket, prefix str
}
out[*o.Key] = et
}
- if resp.IsTruncated == nil || !*resp.IsTruncated || resp.NextContinuationToken == nil {
- return out, nil
- }
- token = resp.NextContinuationToken
}
+ return out, nil
}
-// 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.
+// List routes by memoized bucket→client binding: the interface's caller
+// never says which region a bucket lives in, so the FIRST touch of a bucket
+// probes source-then-destination (a wrong-region List is an error, so the
+// probe is unambiguous) and the winner is remembered.
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
+ m.mu.Lock()
+ c := m.byBucket[bucket]
+ m.mu.Unlock()
+ if c != nil {
+ return m.list(ctx, c, bucket, prefix)
+ }
+ out, err := m.list(ctx, m.src, bucket, prefix)
+ c = m.src
+ if err != nil {
+ if out, err = m.list(ctx, m.dst, bucket, prefix); err != nil {
+ return nil, err
+ }
+ c = m.dst
}
- return m.list(ctx, m.dst, bucket, prefix)
+ m.mu.Lock()
+ m.byBucket[bucket] = c
+ m.mu.Unlock()
+ return out, nil
}
func (m *s3Mirror) Copy(ctx context.Context, srcBucket, srcKey, dstBucket, dstKey string) error {
diff --git a/internal/provision/provision.go b/internal/provision/provision.go
index e5ac03829..cd5f59187 100644
--- a/internal/provision/provision.go
+++ b/internal/provision/provision.go
@@ -651,30 +651,45 @@ func CreateRecord(ctx context.Context, s release.Store, rec Record) error {
// the record is gone, not a manufactured success for a write that touched
// nothing.
func SetRecordRegion(ctx context.Context, s release.Store, acct, app, host, region string) error {
+ _, err := updateRecord(ctx, s, acct, app, host, func(rec *Record) error {
+ rec.Region = region
+ return nil
+ })
+ return err
+}
+
+// updateRecord is the one read-CAS-put-one-retry scaffold under both
+// in-place record writers (SetRecordRegion, SetRecordMove): re-reading and
+// re-applying a narrow mutation on a lost race changes nothing about what
+// it meant. Returns the record as written.
+func updateRecord(ctx context.Context, s release.Store, acct, app, host string, mut func(*Record) error) (*Record, error) {
key := RecordKey(acct, app, host)
- update := func() error {
+ update := func() (*Record, error) {
raw, etag, err := s.Get(ctx, key)
if err != nil {
- return err
+ return nil, 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)
+ return nil, fmt.Errorf("provision: record %s/%s/%s: %w", acct, app, host, err)
+ }
+ if err := mut(&rec); err != nil {
+ return nil, err
}
- rec.Region = region
body, err := json.Marshal(rec)
if err != nil {
- return err
+ return nil, err
}
- _, err = s.Put(ctx, key, body, etag)
- return err
+ if _, err := s.Put(ctx, key, body, etag); err != nil {
+ return nil, err
+ }
+ return &rec, nil
}
- if err := update(); errors.Is(err, release.ErrCASConflict) {
+ rec, err := update()
+ if 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
+ return rec, err
}
// MoveState is the record's in-flight move (issue #331). State is one of
@@ -723,36 +738,16 @@ var ErrMoveConflict = errors.New("provision: move state conflict")
// 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 {
+func SetRecordMove(ctx context.Context, s release.Store, acct, app, host string, mut func(*Record) error) (*Record, error) {
+ return updateRecord(ctx, s, acct, app, host, func(rec *Record) error {
+ if err := mut(rec); err != nil {
return err
}
if rec.Move != nil {
rec.Move.UpdatedAt = time.Now().UTC()
}
- 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
+ return nil
+ })
}
// The five transitions, each a mut for SetRecordMove. They are the ONLY
diff --git a/internal/provision/provision_test.go b/internal/provision/provision_test.go
index a152520b3..37bf85390 100644
--- a/internal/provision/provision_test.go
+++ b/internal/provision/provision_test.go
@@ -899,7 +899,8 @@ func TestSetRecordMoveTransitions(t *testing.T) {
t.Fatal(err)
}
mv := func(mut func(*Record) error) error {
- return SetRecordMove(ctx, s, "bab", "app", "a.go.example.com", mut)
+ _, err := SetRecordMove(ctx, s, "bab", "app", "a.go.example.com", mut)
+ return err
}
read := func() Record {
t.Helper()