amadan

carlos / platform Public

activator: a wake stops when its refreshed route is no longer the one it began for (final fix F4)

An import's rollback lifts its fence before the edge reloads, so a verify
request admitted on the edge's cached import row could wake the host:
doWake refreshed the row to the restored legacy route, never checked it
was still activator-managed, took the host's lease and restored its
replica while the rollback deleted that prefix.

After the refresh, doWake now compares the registry's row with the row
the wake began from (wakeRouteChanged). A row the activator does not
manage, or one with another account, app, address, import gate or seed
id, ends the wake before any lease or restore, as an ordinary failed
wake. Paths moving and a seed's state moving on are not a change.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude dffb1df78b94f6ddc2b79c3d4745c3e91d90d440 parent 173e98d
3 files changed, +175 −0
  • internal/activator/activator.go +5 −0
  • internal/activator/wakeidentity.go +49 −0
  • internal/activator/wakeidentity_test.go +121 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 9e0107012..6d3f4b760 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -1148,6 +1148,11 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
if fresh.DBPath != rt.DBPath || fresh.Socket != rt.Socket {
log.Printf("activator: %s: route refreshed: db %q -> %q socket %q -> %q", rt.Host, rt.DBPath, fresh.DBPath, rt.Socket, fresh.Socket)
}
+ // The refreshed row must still be the route this wake began for,
+ // and the activator's to wake (final review F4, wakeidentity.go).
+ if err := wakeRouteChanged(rt, *fresh); err != nil {
+ return w, err
+ }
rt = *fresh
a.mu.Lock()
if e := a.entries[rt.Host]; e != nil {
diff --git a/internal/activator/wakeidentity.go b/internal/activator/wakeidentity.go
new file mode 100644
index 000000000..811b93b97
--- /dev/null
+++ b/internal/activator/wakeidentity.go
@@ -0,0 +1,49 @@
+package activator
+
+import (
+ "fmt"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/seeds"
+)
+
+// wakeRouteChanged is doWake's check of the row it refreshed against the
+// row the wake began from (final review F4). The edge admits a wake on its
+// cached row, which can be minutes behind the registry: an import's
+// rollback restores the legacy route and lifts its fence before the edge
+// reloads, and a verify request admitted on the cached import row would
+// then take the host's lease and restore its replica while the rollback
+// deletes that same prefix. A refreshed row that the activator no longer
+// manages, or that is now a different route (account, app, seed, import
+// gate or address), is not woken: the answer is an ordinary failed wake,
+// before any lease or restore. nil means the wake goes on.
+func wakeRouteChanged(began, now registry.Route) error {
+ switch {
+ case !ActivatorManaged(now):
+ return fmt.Errorf("%s: the route changed before the wake: the activator no longer manages it", now.Host)
+ case began.Account != now.Account || began.App != now.App:
+ return fmt.Errorf("%s: the route changed before the wake: it belongs to %s/%s now, not %s/%s", now.Host, now.Account, now.App, began.Account, began.App)
+ case began.Addr != now.Addr:
+ return fmt.Errorf("%s: the route changed before the wake: its address is %q now, not %q", now.Host, now.Addr, began.Addr)
+ case began.ImportGate != now.ImportGate:
+ return fmt.Errorf("%s: the route changed before the wake: its import gate changed", now.Host)
+ case rowSeedID(began.SeedSource) != rowSeedID(now.SeedSource):
+ return fmt.Errorf("%s: the route changed before the wake: its seed is %q now, not %q", now.Host, rowSeedID(now.SeedSource), rowSeedID(began.SeedSource))
+ }
+ return nil
+}
+
+// rowSeedID is the seed a row's seed column names: its id, "" for none, and
+// the column itself when it does not parse. A seed's state moves on while
+// it is the same seed (ready, placed, committed), so only the id is the
+// row's identity.
+func rowSeedID(col string) string {
+ rs, err := seeds.ParseRowSeed(col)
+ switch {
+ case err != nil:
+ return col
+ case rs == nil:
+ return ""
+ }
+ return rs.ID
+}
diff --git a/internal/activator/wakeidentity_test.go b/internal/activator/wakeidentity_test.go
new file mode 100644
index 000000000..79caadd7a
--- /dev/null
+++ b/internal/activator/wakeidentity_test.go
@@ -0,0 +1,121 @@
+package activator
+
+import (
+ "context"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/seeds"
+)
+
+// Final review F4: an import's row is cached at the edge; the rollback
+// closes the gate under its fence, puts the legacy address-backed route
+// back and lifts the fence; a verify request then arrives before the edge
+// reloads. The wake refreshes the row, finds the legacy route, and stops:
+// no lease is taken and nothing is restored. The edge is never reloaded.
+func TestAVerifyOnACachedImportRowAfterItsRollbackWakesNothing(t *testing.T) {
+ a, b, _, reg := testActivator(t, time.Hour)
+ a.Routes = reg
+ const host = "old.shop.example.com"
+ if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Addr: "127.0.0.1:8401", App: "shop", Account: "bab"}); err != nil {
+ t.Fatal(err)
+ }
+ original, err := reg.Get(host)
+ if err != nil || original == nil {
+ t.Fatalf("the legacy route: %+v, %v", original, err)
+ }
+ id, err := seeds.NewID()
+ if err != nil {
+ t.Fatal(err)
+ }
+ token := "s3cret-token"
+ imported := registry.Route{Host: host, Kind: "instance", App: "shop", Account: "bab", Provisioned: true, Hibernate: true,
+ Socket: "/run/carlos/instances/old/app.sock", DBPath: "/data/instances/old/" + host + ".db",
+ ImportGate: seeds.HashToken(token), SeedSource: seeds.RowSeed{ID: id, Kind: seeds.KindImport, State: seeds.StateSwapped}.Encode()}
+ if err := reg.SwapForImport(registry.ImportJournal{Host: host, SeedID: id, Original: *original,
+ RecordKey: "apps/bab/shop/instances/" + host + ".json", At: 1}, imported); err != nil {
+ t.Fatal(err)
+ }
+ edge, err := newTestRouter(reg, a) // the edge caches the gated import row
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ // The rollback, steps 1 to 5.
+ const gen = 7
+ if err := reg.CloseImportGate(host, id, gen); err != nil {
+ t.Fatal(err)
+ }
+ fctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ _, _ = a.FenceAndStopIfUnfenced(fctx, host, gen) // nothing is running: the stop is unproven, the fence is held
+ cancel()
+ if a.FenceGeneration(host) != gen {
+ t.Fatal("the rollback's fence was not installed")
+ }
+ if err := reg.RestoreImportRoute(host, id); err != nil {
+ t.Fatal(err)
+ }
+ if !a.LiftFence(host, gen) {
+ t.Fatal("the rollback's fence was not lifted")
+ }
+
+ req := httptest.NewRequest(http.MethodGet, "http://"+host+"/up", nil)
+ req.Header.Set("Carlos-Import-Verify", token)
+ rec := httptest.NewRecorder()
+ edge.ServeHTTP(rec, req)
+ if rec.Code == http.StatusOK {
+ t.Fatalf("the verify was answered %d through the restored legacy route's wake", rec.Code)
+ }
+ b.mu.Lock()
+ acquires, restores := b.acquires, b.restores
+ b.mu.Unlock()
+ if acquires != 0 || restores != 0 {
+ t.Fatalf("the wake took %d leases and ran %d restores for a route the activator no longer manages", acquires, restores)
+ }
+ if got, _ := reg.Get(host); got == nil || got.Addr != original.Addr || got.ImportGate != "" {
+ t.Fatalf("the legacy route after the refused wake = %+v", got)
+ }
+}
+
+// The refreshed row is compared with the row the wake began from: a row
+// the activator does not manage, or one that is another route (account,
+// app, address, import gate, seed), is refused. Paths, versions and a
+// seed's own state moving on are not.
+func TestWakeRouteChanged(t *testing.T) {
+ id, _ := seeds.NewID()
+ other, _ := seeds.NewID()
+ began := testRoute("h.example.com")
+ began.Account, began.App = "bab", "shop"
+ began.SeedSource = seeds.RowSeed{ID: id, Kind: seeds.KindReplica, State: seeds.StateReady}.Encode()
+ for _, c := range []struct {
+ name string
+ edit func(*registry.Route)
+ refused bool
+ }{
+ {"unchanged", func(*registry.Route) {}, false},
+ {"paths moved", func(r *registry.Route) { r.DBPath, r.Socket = "/data/instances/h/h.db", "/run/carlos/instances/h/app.sock" }, false},
+ {"the seed placed", func(r *registry.Route) {
+ r.SeedSource = seeds.RowSeed{ID: id, Kind: seeds.KindReplica, State: seeds.StatePlaced}.Encode()
+ }, false},
+ {"not hibernating", func(r *registry.Route) { r.Hibernate = false }, true},
+ {"no socket", func(r *registry.Route) { r.Socket, r.Addr = "", "127.0.0.1:8401" }, true},
+ {"unit-backed", func(r *registry.Route) { r.Backing = "unit" }, true},
+ {"another account", func(r *registry.Route) { r.Account = "cdef" }, true},
+ {"another app", func(r *registry.Route) { r.App = "blog" }, true},
+ {"an address", func(r *registry.Route) { r.Addr = "127.0.0.1:8401" }, true},
+ {"gated", func(r *registry.Route) { r.ImportGate = registry.ImportGateClosed }, true},
+ {"another seed", func(r *registry.Route) {
+ r.SeedSource = seeds.RowSeed{ID: other, Kind: seeds.KindImport, State: seeds.StateSwapped}.Encode()
+ }, true},
+ {"no seed", func(r *registry.Route) { r.SeedSource = "" }, true},
+ } {
+ now := began
+ c.edit(&now)
+ if err := wakeRouteChanged(began, now); (err != nil) != c.refused {
+ t.Errorf("%s: %v, want refused %v", c.name, err, c.refused)
+ }
+ }
+}