amadan

carlos / platform Public

activator: staging-fallback restore, the full local purge, Quiesced (#331)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PEypt7de2X1G84aQ9rNkCf
Paul Campbell pushed by paul@keymail.dev 296d8ac708fed5c306405e451c0ed14fa996355a parent 2824919
3 files changed, +222 −4
  • internal/activator/activator.go +12 −0
  • internal/activator/backend_exec.go +94 −4
  • internal/activator/backend_exec_test.go +116 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 2ec94c275..609ccf743 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -1080,6 +1080,18 @@ func (a *Service) RetireExcept(keep map[string]bool) {
}
}
+// Quiesced answers whether a host currently has NO live activator entry —
+// no child running, none waking, none mid-stop. The residence-move cut
+// (internal/host's move worker) gates its final mirror round on this: the
+// route's Status change retires the entry through RetireExcept, and only a
+// quiet host's replica prefix is byte-final.
+func (a *Service) Quiesced(host string) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ _, live := a.entries[host]
+ return !live
+}
+
// Shutdown checkpoints and stops every child — the agent going down takes its
// instances with it (they're its children; leaving them orphaned would strand
// the leases until TTL expiry and leave the sockets bound against the next
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index a8fac9f93..2903a7c35 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -60,6 +60,14 @@ type ExecBackend struct {
// but a carlos app is a separate binary from the release store — there is
// nothing sensible to invent here, and the adopt pass owns the symlinks.
Materialize func(ctx context.Context, host string) error
+ // Mirror, when non-nil, arms Restore's staging fallback (issue #331):
+ // a host with no replica under the ordinary instances/ prefix may have
+ // one under moves/<host>/ — a residence move's staged copy, pushed by
+ // the SOURCE box. Restore then copies it into place (own bucket, own
+ // IAM), restores as normal, and deletes the staging copy once CheckDB
+ // vouches for what it restored. Nil = the fallback does not exist,
+ // byte-identical to the pre-move behaviour.
+ Mirror lease.MirrorStore
// Region is the box's CARLOS placement code (CARLOS_REGION), handed to
// every exec child as its own CARLOS_REGION (issue #329, titogo's
// residence design: the app reads where it lives — e.g. to clear a move
@@ -226,11 +234,22 @@ func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, err
// its MetaDirSuffix — spelled out because this package never imports
// litestream) as any box-wide backup replicator would use, and the lease
// package's activator-scoped ".activator-<db>-litestream".
- for _, p := range []string{dbPath + "-wal", dbPath + "-shm"} {
- _ = os.Remove(p)
+ PurgeLocalDBArtifacts(dbPath, false)
+
+ // The staging fallback (issue #331): when the ordinary prefix is empty
+ // and a move staged a replica here, promote it — server-side, own
+ // bucket — and restore from the promoted copy. Precedence is the
+ // ordinary prefix ALWAYS: a host that re-lands here after moving away
+ // and back must restore its own lineage, never regress to an old
+ // staged copy (the source's cut deleted the own prefix it left behind,
+ // so a non-empty own prefix is by construction the newer story).
+ promoted := false
+ if b.Mirror != nil {
+ var err error
+ if promoted, err = b.promoteStagedReplica(ctx, host); err != nil {
+ log.Printf("activator: %s: staging fallback: %v (continuing without it)", host, err)
+ }
}
- _ = os.RemoveAll(dbPath + "-litestream")
- _ = os.RemoveAll(filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream"))
restored, err := restoreReplica(ctx, dbPath, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region)
if err != nil {
@@ -270,6 +289,14 @@ func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, err
return false, fmt.Errorf("restored replica for %s is corrupt: %w", host, err)
}
log.Printf("activator: restored %s from s3://%s/%s", host, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host))
+ // A clean restore of a copy promoted THIS call is the moment the
+ // staging has served its purpose; a corrupt one (above)
+ // deliberately LEFT it in place as evidence, and staging that was
+ // never consulted (an abort's leftovers under a live own lineage)
+ // is not this restore's to touch.
+ if promoted {
+ b.dropStagedReplica(ctx, host)
+ }
} else {
// Never replicated: a brand-new instance boots fresh and creates its
// own DB — the register-only create path (provisioning as a metadata
@@ -999,3 +1026,66 @@ func (p *childProc) Stop(ctx context.Context) error {
return fmt.Errorf("instance did not exit on SIGTERM; killed")
}
}
+
+// movesPrefix is where a residence move stages a replica in the TARGET's
+// data bucket (instance-move spec): moves/<host>/ beside the ordinary
+// instances/ prefix. A free function so the host's move worker and this
+// package spell it identically.
+func MovesPrefix(host string) string { return "moves/" + host + "/" }
+
+// PurgeLocalDBArtifacts removes the local litestream footprint next to a
+// DB — WAL, SHM and both meta dirs — and, when full is true, the DB file
+// itself. Restore's pre-restore cleanup uses the partial form (the DB's
+// absence is what got us here); a residence move's cut uses the full form:
+// the activator's warm-cache trust is exactly right for crashes and exactly
+// wrong after a cut, where a stale local file would resurrect a moved-away
+// instance's history.
+func PurgeLocalDBArtifacts(dbPath string, full bool) {
+ for _, p := range []string{dbPath + "-wal", dbPath + "-shm"} {
+ _ = os.Remove(p)
+ }
+ _ = os.RemoveAll(dbPath + "-litestream")
+ _ = os.RemoveAll(filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream"))
+ if full {
+ _ = os.Remove(dbPath)
+ }
+}
+
+// promoteStagedReplica copies moves/<host>/ over instances/<host>/ in this
+// box's OWN bucket when — and only when — the ordinary prefix is empty. A
+// no-op (nil error) when there is nothing staged.
+func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (bool, error) {
+ own, err := b.Mirror.List(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host))
+ if err != nil {
+ return false, err
+ }
+ if len(own) > 0 {
+ return false, nil // the ordinary lineage exists; staging never wins
+ }
+ staged, err := b.Mirror.List(ctx, b.Cfg.Bucket, MovesPrefix(host))
+ if err != nil {
+ return false, err
+ }
+ if len(staged) == 0 {
+ return false, nil
+ }
+ log.Printf("activator: %s: promoting %d staged object(s) from %s (residence move landing)", host, len(staged), MovesPrefix(host))
+ if _, err := lease.MirrorPrefix(ctx, b.Mirror, b.Cfg.Bucket, MovesPrefix(host), b.Cfg.Bucket, b.Cfg.hibernatePrefix(host)); err != nil {
+ return false, err
+ }
+ return true, nil
+}
+
+// dropStagedReplica deletes moves/<host>/ after a restore CheckDB vouched
+// for. Best-effort: a leftover staging copy is storage, not a correctness
+// problem — promoteStagedReplica never consults it again once the ordinary
+// prefix exists.
+func (b *ExecBackend) dropStagedReplica(ctx context.Context, host string) {
+ staged, err := b.Mirror.List(ctx, b.Cfg.Bucket, MovesPrefix(host))
+ if err != nil {
+ return
+ }
+ for k := range staged {
+ _ = b.Mirror.Delete(ctx, b.Cfg.Bucket, k)
+ }
+}
diff --git a/internal/activator/backend_exec_test.go b/internal/activator/backend_exec_test.go
index 7e3fb3a8a..3c14ea294 100644
--- a/internal/activator/backend_exec_test.go
+++ b/internal/activator/backend_exec_test.go
@@ -2,6 +2,7 @@ package activator
import (
"context"
+ "database/sql"
"os"
"os/exec"
"path/filepath"
@@ -9,6 +10,8 @@ import (
"testing"
"time"
+ _ "modernc.org/sqlite"
+
"github.com/carlosframework/platform/internal/cgroup"
"github.com/carlosframework/platform/internal/registry"
)
@@ -103,3 +106,116 @@ func TestExecBackendStartChildRegion(t *testing.T) {
})
}
}
+
+// The staging fallback (issue #331): a host whose ordinary prefix is empty
+// restores from a move's staged copy — promoted server-side, dropped after
+// a clean restore — and the ordinary prefix ALWAYS wins when it exists.
+func TestRestoreStagingFallback(t *testing.T) {
+ ctx := context.Background()
+ b := testExecBackend(t)
+ f := newFakeMirrorStore()
+ b.Mirror = f
+
+ host := "moved.example.com"
+ rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(t.TempDir(), "i.sock")}
+
+ // restoreReplica is stubbed for the WHOLE test (the real one dials S3):
+ // it restores — materializing a genuinely valid SQLite file, so CheckDB
+ // passes — exactly when the fake bucket holds objects under the prefix
+ // it was pointed at, which is the promotion's observable effect.
+ var restoredFrom []string
+ oldRestore := restoreReplica
+ restoreReplica = func(_ context.Context, dbPath, bucket, prefix, region string) (bool, error) {
+ objs, _ := f.List(context.Background(), bucket, prefix)
+ if len(objs) == 0 {
+ return false, nil
+ }
+ restoredFrom = append(restoredFrom, bucket+"/"+prefix)
+ return true, writeMinimalSQLite(dbPath)
+ }
+ defer func() { restoreReplica = oldRestore }()
+
+ // Nothing staged, nothing own: plain fresh boot (restored=false→fresh=true).
+ fresh, err := b.Restore(ctx, rt)
+ if err != nil || !fresh {
+ t.Fatalf("empty everything: fresh=%v err=%v, want fresh boot", fresh, err)
+ }
+
+ // Staged copy present, own prefix empty: promoted then restored.
+ f.put(b.Cfg.Bucket, "moves/"+host+"/generations/a/snapshot", "e1")
+ _ = os.Remove(filepath.Join(b.DataDir, host+".db"))
+ fresh, err = b.Restore(ctx, rt)
+ if err != nil {
+ t.Fatalf("staged restore: %v", err)
+ }
+ if fresh {
+ t.Fatal("staged restore reported a fresh boot")
+ }
+ if len(restoredFrom) != 1 || restoredFrom[0] != b.Cfg.Bucket+"/"+b.Cfg.hibernatePrefix(host) {
+ t.Fatalf("restored from %v, want the promoted ordinary prefix %s", restoredFrom, b.Cfg.hibernatePrefix(host))
+ }
+ own, _ := f.List(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host))
+ if len(own) != 1 {
+ t.Fatalf("promotion left own prefix = %v", own)
+ }
+ staged, _ := f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/")
+ if len(staged) != 0 {
+ t.Fatalf("staging not dropped after a clean restore: %v", staged)
+ }
+
+ // Precedence: own prefix present → staging never consulted, never
+ // dropped.
+ f.put(b.Cfg.Bucket, "moves/"+host+"/generations/b/snapshot", "e9")
+ _ = os.Remove(filepath.Join(b.DataDir, host+".db"))
+ restoredFrom = nil
+ if _, err := b.Restore(ctx, rt); err != nil {
+ t.Fatal(err)
+ }
+ staged, _ = f.List(ctx, b.Cfg.Bucket, "moves/"+host+"/")
+ if len(staged) != 1 {
+ t.Fatalf("staging touched despite an existing own lineage: %v", staged)
+ }
+}
+
+// fakeMirrorStore is the test double for lease.MirrorStore, bucket→key→etag.
+type fakeMirrorStore struct{ buckets map[string]map[string]string }
+
+func newFakeMirrorStore() *fakeMirrorStore {
+ return &fakeMirrorStore{buckets: map[string]map[string]string{}}
+}
+func (f *fakeMirrorStore) put(bucket, key, etag string) {
+ if f.buckets[bucket] == nil {
+ f.buckets[bucket] = map[string]string{}
+ }
+ f.buckets[bucket][key] = etag
+}
+func (f *fakeMirrorStore) List(_ context.Context, bucket, prefix string) (map[string]string, error) {
+ out := map[string]string{}
+ for k, v := range f.buckets[bucket] {
+ if strings.HasPrefix(k, prefix) {
+ out[k] = v
+ }
+ }
+ return out, nil
+}
+func (f *fakeMirrorStore) Copy(_ context.Context, srcBucket, srcKey, dstBucket, dstKey string) error {
+ f.put(dstBucket, dstKey, f.buckets[srcBucket][srcKey])
+ return nil
+}
+func (f *fakeMirrorStore) Delete(_ context.Context, bucket, key string) error {
+ delete(f.buckets[bucket], key)
+ return nil
+}
+
+// writeMinimalSQLite creates a genuinely valid empty SQLite database so
+// lease.CheckDB passes on the stubbed restore's output.
+func writeMinimalSQLite(dbPath string) error {
+ db, err := sql.Open("sqlite", dbPath)
+ if err != nil {
+ return err
+ }
+ if _, err := db.Exec("create table t(x)"); err != nil {
+ return err
+ }
+ return db.Close()
+}