amadan

carlos / platform Public

host: wire the import swap and rollback; fence generations survive restarts (Plan 7 Task 7)

wireSeedPlacement (preflight M7) now arms, with claim control off, the
swap's NestedOK and LegacyOwners and all six rollback hooks together:
fenceForRollback over the activator (the layout pass's rule, typed by
layout.Fence, m13), LiftFence, RemoveLocal as activator.PurgeSeedLocal,
LocalSeedID as activator.ReadSeedMarker, deleteReplica under
<prefix>/<host>/ and NextFenceGen. agent.go passes the activator,
cfg.Prefix, the startup-sweep flag and the generation source; the
recovery closure records the sweep. Claim-armed boxes arm only Data.

Fence generations come from the registry (fence_generation): the later
of the wall clock and one past the last handed out, stored before use,
so none repeats across an agent restart or a clock step back. The layout
pass draws from the same source. NextFenceGen returns an error, and a
failed or zero generation closes nothing; fenceForRollback refuses gen 0.

seeds.ProtocolVersion stays 0 (M9): Plan 8's last task raises it.
TestWithSeedsFields asserts seeds.ProtocolVersion (M8); a test pins
seeds.ClosedGate to registry.ImportGateClosed.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 274bdeca97996b9ef0397d149ba5b909b49d8d8a parent b7754cb
15 files changed, +727 −21
  • internal/host/agent.go +6 −2
  • internal/host/layoutleg.go +14 −0
  • internal/host/seedsfloor_linux_test.go +3 −2
  • internal/host/seedwiring.go +121 −2
  • internal/host/seedwiring_linux_test.go +63 −0
  • internal/host/seedwiring_test.go +303 −3
  • internal/provision/seedreconcile.go +5 −2
  • internal/provision/seedrollback.go +9 −1
  • internal/provision/seedrollback_test.go +30 −3
  • internal/registry/fencegen.go +52 −0
  • internal/registry/fencegen_test.go +90 −0
  • internal/registry/registry.go +3 −0
  • internal/seeds/floor.go +11 −3
  • internal/seeds/floor_test.go +6 −3
  • internal/seeds/seeds_test.go +11 −0

CI passed — run details

diff --git a/internal/host/agent.go b/internal/host/agent.go
index a9959a147..80f3c1d29 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -410,9 +410,12 @@ func runAgent(args []string, buildClaims ClaimFactory) {
// Seeded instances (seeded-instances spec §3.4): the seed reader,
// the placed recorder, the provision pass's hooks and the claim
// flag the seed fallback refuses under. Before Start, like the
- // mirror.
+ // mirror. The import swap and rollback (spec §5.4, §5.6) fence
+ // through this activator, delete the replica under cfg.Prefix, and
+ // trust an unproven stop only after the startup sweep below.
wireSeedPlacement(eb, claims.Writes.Enabled, seedDataOpener(cfg.Bucket, cfg.Region),
- func() (release.Store, error) { return openDeployStore(context.Background()) })
+ func() (release.Store, error) { return openDeployStore(context.Background()) },
+ seedBox{Fence: svc, Prefix: cfg.Prefix, Swept: seedSweptAtStart.Load, NextFenceGen: fenceGenerations(reg)})
armMovesPass(*cfg, dd, reg, svc, eb, claims.Writes)
armRunsPass(svc, reg, dd)
// The startup sweep (design §6.7): a leftover instance unit from a
@@ -442,6 +445,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {
return err
},
func(ctx context.Context, swept bool) {
+ seedSweptAtStart.Store(swept)
pass := newLayoutPass(reg, svc, dd, eb.NestedOK, swept)
intent := startupRecoveryIntent(ctx, dd, openDeployStore, metricsBoxLabel(os.Getenv, readMachineID()))
recoverLayoutAtStart(ctx, pass, intent)
diff --git a/internal/host/layoutleg.go b/internal/host/layoutleg.go
index a035f412b..396d3bacf 100644
--- a/internal/host/layoutleg.go
+++ b/internal/host/layoutleg.go
@@ -325,6 +325,20 @@ func newLayoutPass(reg *registry.Registry, svc *activator.Service, dd string, ne
if svc != nil {
p.Fence = svc
}
+ // The pass fences from the box's generation source, the one the import
+ // rollback's journal stores its generation from (fenceGenerations), so
+ // a rollback resumed after a restart never meets this pass's fence
+ // under its own number. A source that fails falls back to the pass's
+ // own default, the wall clock.
+ if reg != nil {
+ next := fenceGenerations(reg)
+ p.Generation = func() uint64 {
+ if g, err := next(); err == nil {
+ return g
+ }
+ return uint64(time.Now().UnixNano())
+ }
+ }
// The default's readiness is this same gate (defaultLayoutIntent).
if nestedOK != nil {
layoutDefaultGate.Store(&nestedOK)
diff --git a/internal/host/seedsfloor_linux_test.go b/internal/host/seedsfloor_linux_test.go
index a85a068c8..59a4b84eb 100644
--- a/internal/host/seedsfloor_linux_test.go
+++ b/internal/host/seedsfloor_linux_test.go
@@ -5,6 +5,7 @@ package host
import (
"context"
"encoding/json"
+ "fmt"
"os"
"path/filepath"
"strings"
@@ -138,7 +139,7 @@ func TestWithSeedsFields(t *testing.T) {
t.Fatalf("armed = %+v", f)
}
body, _ := json.Marshal(f)
- if !strings.Contains(string(body), `"seeds_protocol":0`) {
- t.Fatalf("seeds_protocol must be on the wire even at 0: %s", body)
+ if want := fmt.Sprintf(`"seeds_protocol":%d`, seeds.ProtocolVersion); !strings.Contains(string(body), want) {
+ t.Fatalf("seeds_protocol must be on the wire, even at 0: want %s in %s", want, body)
}
}
diff --git a/internal/host/seedwiring.go b/internal/host/seedwiring.go
index f4dca6af7..85faac55e 100644
--- a/internal/host/seedwiring.go
+++ b/internal/host/seedwiring.go
@@ -8,9 +8,13 @@ import (
"log"
"os"
"sync/atomic"
+ "time"
"github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/controlname"
+ "github.com/carlosframework/platform/internal/layout"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
"github.com/carlosframework/platform/internal/seeds"
)
@@ -103,6 +107,17 @@ func seedDataOpener(bucket, region string) func() (release.Store, error) {
}
}
+// seedBox is what the import swap and rollback need from this box beyond
+// the two stores (preflight M7): the activator's fence, the key prefix of
+// the replicas in the data bucket (cfg.Prefix), whether the startup unit
+// sweep succeeded, and the fence generation source.
+type seedBox struct {
+ Fence layout.Fence
+ Prefix string
+ Swept func() bool
+ NextFenceGen func() (uint64, error)
+}
+
// wireSeedPlacement arms seeded instances on this box (seeded-instances
// spec §3.4): the claim flag that makes the seed fallback refuse, the data
// bucket for snapshots, the record store for the placed CAS, and the
@@ -110,7 +125,19 @@ func seedDataOpener(bucket, region string) func() (release.Store, error) {
// rest is best-effort like the staging-fallback mirror, and all or nothing:
// a box that cannot build both stores places no seed, and a seeded row's
// wake fails instead of starting empty.
-func wireSeedPlacement(eb *activator.ExecBackend, claimArmed bool, openData, openDeploy func() (release.Store, error)) {
+//
+// The import swap's and rollback's hooks (spec §5.4, §5.6) are armed only
+// with claim control off: a claim-armed box refuses seeds until #250, so it
+// never swaps, and an import journaled before claims came on stays gated
+// and untouched there. The rollback's six hooks are armed together or not
+// at all (provision runs no rollback missing one); without a fence from the
+// activator none is.
+//
+// The rollback's fence is known by the generation its journal stores, so
+// box.NextFenceGen must never repeat one across a restart
+// (fenceGenerations). The hooks set no clock: the swap and the rollback
+// read the pass's own (ReconcileConfig.Now).
+func wireSeedPlacement(eb *activator.ExecBackend, claimArmed bool, openData, openDeploy func() (release.Store, error), box seedBox) {
eb.ClaimArmed = claimArmed
data, err := openData()
if err != nil {
@@ -124,5 +151,97 @@ func wireSeedPlacement(eb *activator.ExecBackend, claimArmed bool, openData, ope
}
eb.Seeds = storeSeedObjects{data}
eb.OnSeedPlaced = seedPlacedHook(deploy, data)
- armSeedHooks(&provision.SeedHooks{Data: data})
+ h := &provision.SeedHooks{Data: data}
+ if !claimArmed {
+ h.NestedOK = eb.NestedOK
+ h.LegacyOwners = func(ctx context.Context) (*seeds.LegacyOwners, error) {
+ o, _, err := seeds.ReadLegacyOwners(ctx, deploy)
+ return o, err
+ }
+ if box.Fence != nil && box.Swept != nil && box.NextFenceGen != nil {
+ fence := box.Fence
+ h.FenceForRollback = fenceForRollback(fence, box.Swept)
+ h.LiftFence = func(host string, gen uint64) { fence.LiftFence(host, gen) }
+ h.RemoveLocal = func(rt registry.Route) error { return activator.PurgeSeedLocal(rt.DBPath) }
+ h.LocalSeedID = activator.ReadSeedMarker
+ h.DeleteReplica = deleteReplica(data, box.Prefix)
+ h.NextFenceGen = box.NextFenceGen
+ }
+ }
+ armSeedHooks(h)
+}
+
+// seedSweptAtStart is whether the agent's startup unit sweep succeeded,
+// for the import rollback's stop proof: the same fact the layout pass is
+// built with. It stays false with claim control on, where the activator
+// starts without a sweep (and no rollback is armed).
+var seedSweptAtStart atomic.Bool
+
+// fenceGenerations is this box's fence generation source, the registry's:
+// never 0, and never one it handed out before, across agent restarts. The
+// layout pass draws from it too (newLayoutPass), so a rollback resumed
+// after a restart can never meet the pass's fence under its own number.
+func fenceGenerations(reg *registry.Registry) func() (uint64, error) {
+ return func() (uint64, error) { return reg.NextFenceGeneration(time.Now()) }
+}
+
+// rollbackFenceTimeout bounds one fence-and-stop, as the layout pass's.
+const rollbackFenceTimeout = 30 * time.Second
+
+// fenceForRollback is the import rollback's step 2 (seeded-instances spec
+// §5.6), by internal/layout/pass.go's rule (Pass.quiesce) for the same
+// activator API. A fence already holding gen is this rollback's own, from
+// an earlier pass of this process: proceed. Otherwise fence and stop under
+// a 30-second deadline. A proven stop proceeds; another operation's fence
+// defers, touching nothing; an unproven stop proceeds only when the box
+// was swept at start, the host is quiesced and its stop is not uncertain.
+// Anything else lifts this attempt's fence and defers.
+//
+// Generation 0 is no fence's: FenceGeneration answers 0 for a host with
+// none, which must never read as ours.
+func fenceForRollback(f layout.Fence, swept func() bool) func(ctx context.Context, host string, gen uint64) (bool, error) {
+ return func(ctx context.Context, host string, gen uint64) (bool, error) {
+ if gen == 0 {
+ return false, activator.ErrFenceGeneration
+ }
+ if f.FenceGeneration(host) == gen {
+ return true, nil
+ }
+ fctx, cancel := context.WithTimeout(ctx, rollbackFenceTimeout)
+ defer cancel()
+ _, err := f.FenceAndStopIfUnfenced(fctx, host, gen)
+ switch {
+ case err == nil:
+ return true, nil
+ case errors.Is(err, activator.ErrFenceHeld):
+ return false, err
+ case errors.Is(err, activator.ErrStopUnproven) && swept() && f.Quiesced(host) && !f.StopUncertain(host):
+ return true, nil
+ }
+ f.LiftFence(host, gen)
+ return false, err
+ }
+}
+
+// deleteReplica is the rollback's step 6 for the host's replica: every key
+// under <prefix>/<host>/ in the region data bucket, the prefix the
+// activator replicates to (activator.HibernatePrefix). The trailing slash
+// keeps a host whose name only begins with this one out of it. Nothing
+// there is not an error.
+func deleteReplica(data release.Store, prefix string) func(ctx context.Context, host string) error {
+ return func(ctx context.Context, host string) error {
+ if !controlname.ValidHostShape(host) {
+ return fmt.Errorf("delete the replica of %q: not a host name", host)
+ }
+ keys, err := data.List(ctx, activator.HibernatePrefix(prefix, host)+"/")
+ if err != nil {
+ return err
+ }
+ for _, k := range keys {
+ if err := data.Delete(ctx, k); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ }
+ return nil
+ }
}
diff --git a/internal/host/seedwiring_linux_test.go b/internal/host/seedwiring_linux_test.go
new file mode 100644
index 000000000..a316f2c8d
--- /dev/null
+++ b/internal/host/seedwiring_linux_test.go
@@ -0,0 +1,63 @@
+//go:build linux
+
+package host
+
+import (
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The wired LocalSeedID and RemoveLocal are Plan 4's marker read and purge
+// (preflight M12), through instancefs: the marker names the seed, and the
+// removal takes the database, its WAL and SHM, the marker and the seed
+// fetch leftovers, and leaves anything else in the directory.
+func TestWiredRollbackReadsAndRemovesTheLocalSeed(t *testing.T) {
+ h := wiredSeedHooks(t, &fakeFence{gens: map[string]uint64{}}, restartTestRegistry(t),
+ release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir()))
+ const id = "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa"
+ dir := filepath.Join(t.TempDir(), "shop-1.example.com")
+ if err := os.MkdirAll(dir, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ db := filepath.Join(dir, "shop-1.example.com.db")
+ gone := []string{db, db + "-wal", db + "-shm", db + activator.SeedMarkerSuffix, filepath.Join(dir, ".seed-sd_x.tmp")}
+ for _, p := range gone {
+ body := "data"
+ if p == db+activator.SeedMarkerSuffix {
+ body = id + "\n"
+ }
+ if err := os.WriteFile(p, []byte(body), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ kept := filepath.Join(dir, "uploads.txt")
+ if err := os.WriteFile(kept, []byte("keep"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+
+ if got, err := h.LocalSeedID(db); err != nil || got != id {
+ t.Fatalf("LocalSeedID = %q, %v; want %q", got, err, id)
+ }
+ if err := h.RemoveLocal(registry.Route{Host: "shop-1.example.com", Kind: "instance", DBPath: db}); err != nil {
+ t.Fatal(err)
+ }
+ for _, p := range gone {
+ if _, err := os.Lstat(p); !os.IsNotExist(err) {
+ t.Errorf("%s survived the removal (%v)", p, err)
+ }
+ }
+ if _, err := os.Stat(kept); err != nil {
+ t.Errorf("an unrelated file went: %v", err)
+ }
+ if got, err := h.LocalSeedID(db); err != nil || got != "" {
+ t.Fatalf("LocalSeedID after the removal = %q, %v", got, err)
+ }
+ if err := h.RemoveLocal(registry.Route{Host: "shop-1.example.com", Kind: "instance", DBPath: db}); err != nil {
+ t.Fatalf("a second removal: %v", err)
+ }
+}
diff --git a/internal/host/seedwiring_test.go b/internal/host/seedwiring_test.go
index 010bf1be8..aad57d368 100644
--- a/internal/host/seedwiring_test.go
+++ b/internal/host/seedwiring_test.go
@@ -5,13 +5,16 @@ import (
"errors"
"io"
"os"
+ "path/filepath"
"sync"
"testing"
"time"
"github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/domains"
+ "github.com/carlosframework/platform/internal/layout"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
"github.com/carlosframework/platform/internal/seeds"
)
@@ -250,7 +253,7 @@ func TestWireSeedPlacement(t *testing.T) {
t.Run("armed", func(t *testing.T) {
resetArmedSeedHooks(t)
eb := &activator.ExecBackend{}
- wireSeedPlacement(eb, true, ok(data), ok(deploy))
+ wireSeedPlacement(eb, true, ok(data), ok(deploy), seedBox{})
if !eb.ClaimArmed || eb.Seeds == nil || eb.OnSeedPlaced == nil {
t.Fatalf("ClaimArmed %v Seeds %v OnSeedPlaced set %v", eb.ClaimArmed, eb.Seeds, eb.OnSeedPlaced != nil)
}
@@ -262,7 +265,7 @@ func TestWireSeedPlacement(t *testing.T) {
t.Run("claims off", func(t *testing.T) {
resetArmedSeedHooks(t)
eb := &activator.ExecBackend{ClaimArmed: true}
- wireSeedPlacement(eb, false, ok(data), ok(deploy))
+ wireSeedPlacement(eb, false, ok(data), ok(deploy), seedBox{})
if eb.ClaimArmed {
t.Fatal("ClaimArmed set with claim control off")
}
@@ -274,7 +277,7 @@ func TestWireSeedPlacement(t *testing.T) {
t.Run(name, func(t *testing.T) {
resetArmedSeedHooks(t)
eb := &activator.ExecBackend{}
- wireSeedPlacement(eb, true, opens[0], opens[1])
+ wireSeedPlacement(eb, true, opens[0], opens[1], seedBox{})
if !eb.ClaimArmed {
t.Fatal("ClaimArmed not set when a store failed")
}
@@ -320,3 +323,300 @@ func TestReconcileSerializedCarriesTheArmedSeedHooks(t *testing.T) {
t.Fatal("the aborted seed's snapshot survived an armed pass")
}
}
+
+// fakeFence is the activator's fence as the import rollback meets it: one
+// fence per host, an exclusive install that refuses any fence already
+// held, and a lift that takes only the generation it names.
+// FenceAndStopIfUnfenced answers err once it would install; nil and
+// ErrStopUnproven leave the fence installed, as the activator does.
+type fakeFence struct {
+ gens map[string]uint64
+ err error
+ quiesced bool
+ uncertain bool
+ calls int
+ deadline time.Duration // the time left on the last install's context
+ lifted []uint64
+}
+
+func (f *fakeFence) FenceAndStopIfUnfenced(ctx context.Context, host string, gen uint64) (*activator.StopEvidence, error) {
+ f.calls++
+ if d, ok := ctx.Deadline(); ok {
+ f.deadline = time.Until(d)
+ }
+ if f.gens[host] != 0 {
+ return nil, activator.ErrFenceHeld
+ }
+ if f.err == nil || errors.Is(f.err, activator.ErrStopUnproven) {
+ f.gens[host] = gen
+ }
+ return nil, f.err
+}
+
+func (f *fakeFence) LiftFence(host string, gen uint64) bool {
+ f.lifted = append(f.lifted, gen)
+ if gen == 0 || f.gens[host] != gen {
+ return false
+ }
+ delete(f.gens, host)
+ return true
+}
+func (f *fakeFence) FenceGeneration(host string) uint64 { return f.gens[host] }
+func (f *fakeFence) Quiesced(string) bool { return f.quiesced }
+func (f *fakeFence) StopUncertain(string) bool { return f.uncertain }
+
+func storeOpener(s release.Store) func() (release.Store, error) {
+ return func() (release.Store, error) { return s, nil }
+}
+
+// wiredSeedHooks arms seed placement as the agent does on a box without
+// claim control, with f as the activator and reg's generation source, and
+// returns the provision pass's hooks.
+func wiredSeedHooks(t *testing.T, f layout.Fence, reg *registry.Registry, data, deploy release.Store) *provision.SeedHooks {
+ t.Helper()
+ resetArmedSeedHooks(t)
+ eb := &activator.ExecBackend{NestedOK: func() error { return nil }}
+ wireSeedPlacement(eb, false, storeOpener(data), storeOpener(deploy), seedBox{
+ Fence: f, Prefix: "instances", Swept: func() bool { return true }, NextFenceGen: fenceGenerations(reg)})
+ h := armedSeedHooks.Load()
+ if h == nil {
+ t.Fatal("no seed hooks armed")
+ }
+ return h
+}
+
+// The rollback's step 2 (spec §5.6) by the layout pass's rule, and the
+// generation it is known by surviving an agent restart.
+func TestRollbackFenceAfterRestart(t *testing.T) {
+ ctx := context.Background()
+ const host = "shop-1.example.com"
+ swept := true
+ f := &fakeFence{gens: map[string]uint64{}, err: activator.ErrStopUnproven, quiesced: true}
+ fence := fenceForRollback(f, func() bool { return swept })
+ if ok, _ := fence(ctx, host, 7); !ok {
+ t.Fatal("cold, swept, quiesced: want proceed")
+ }
+ if f.deadline <= 0 || f.deadline > 30*time.Second {
+ t.Fatalf("the install ran with %v left, want a deadline of at most 30s", f.deadline)
+ }
+ // Ours already (an earlier pass installed it): proceed without asking.
+ f.err, f.calls = errors.New("must not be called"), 0
+ if ok, _ := fence(ctx, host, 7); !ok || f.calls != 0 {
+ t.Fatalf("our own fence: want proceed without an install (calls %d)", f.calls)
+ }
+ // Not swept at start: lift and defer.
+ f = &fakeFence{gens: map[string]uint64{}, err: activator.ErrStopUnproven, quiesced: true}
+ swept = false
+ fence = fenceForRollback(f, func() bool { return swept })
+ if ok, _ := fence(ctx, host, 8); ok || len(f.lifted) != 1 || f.lifted[0] != 8 || f.gens[host] != 0 {
+ t.Fatalf("not swept: proceed %v lifted %v fences %v", ok, f.lifted, f.gens)
+ }
+ // Swept, but not quiesced, or its stop uncertain: lift and defer.
+ for name, f := range map[string]*fakeFence{
+ "not quiesced": {gens: map[string]uint64{}, err: activator.ErrStopUnproven},
+ "stop uncertain": {gens: map[string]uint64{}, err: activator.ErrStopUnproven, quiesced: true, uncertain: true},
+ "another error": {gens: map[string]uint64{}, err: activator.ErrFenceNotInstalled, quiesced: true},
+ } {
+ fence := fenceForRollback(f, func() bool { return true })
+ if ok, err := fence(ctx, host, 9); ok || err == nil || f.gens[host] != 0 {
+ t.Fatalf("%s: proceed %v, %v, fences %v", name, ok, err, f.gens)
+ }
+ }
+ // Another operation's fence: defer and touch nothing.
+ f = &fakeFence{gens: map[string]uint64{host: 3}}
+ fence = fenceForRollback(f, func() bool { return true })
+ if ok, err := fence(ctx, host, 9); ok || !errors.Is(err, activator.ErrFenceHeld) || len(f.lifted) != 0 || f.gens[host] != 3 {
+ t.Fatalf("held: proceed %v, %v, lifted %v", ok, err, f.lifted)
+ }
+ // Generation 0 is no fence's: on a host with none, FenceGeneration's 0
+ // would otherwise read as "ours already" and proceed unfenced.
+ f = &fakeFence{gens: map[string]uint64{}}
+ fence = fenceForRollback(f, func() bool { return true })
+ if ok, err := fence(ctx, host, 0); ok || err == nil || f.calls != 0 {
+ t.Fatalf("generation 0: proceed %v, %v, installs %d", ok, err, f.calls)
+ }
+
+ // After an agent restart. The journal kept the generation the rollback
+ // was given before it; the activator's fences died with the process,
+ // and another operation, the layout pass, has fenced the host since.
+ // The resumed rollback must neither take that fence for its own nor
+ // lift it: provision lifts at the journal's generation on every path
+ // that does not proceed, the held-stale-journal lift
+ // (seedrollback.go, rollBackRoute's first check) among them.
+ path := filepath.Join(t.TempDir(), "registry.db")
+ reg, err := registry.Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // The clock before the restart ran an hour ahead of the clock after it
+ // (a box that booted before NTP synced): the generation the journal
+ // stores is ahead of every wall-clock reading the restarted agent makes.
+ if _, err := reg.NextFenceGeneration(time.Now().Add(time.Hour)); err != nil {
+ t.Fatal(err)
+ }
+ data, deploy := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ before := wiredSeedHooks(t, &fakeFence{gens: map[string]uint64{}}, reg, data, deploy)
+ stored, err := before.NextFenceGen()
+ if err != nil || stored == 0 {
+ t.Fatalf("generation before the restart: %d, %v", stored, err)
+ }
+ if err := reg.Close(); err != nil {
+ t.Fatal(err)
+ }
+
+ reg, err = registry.Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer reg.Close()
+ other := newLayoutPass(reg, nil, t.TempDir(), nil, true).Generation()
+ if other <= stored {
+ t.Fatalf("the layout pass after the restart drew generation %d, not above the %d the journal holds", other, stored)
+ }
+ f = &fakeFence{gens: map[string]uint64{host: other}, err: activator.ErrStopUnproven, quiesced: true}
+ after := wiredSeedHooks(t, f, reg, data, deploy)
+ if ok, err := after.FenceForRollback(ctx, host, stored); ok || !errors.Is(err, activator.ErrFenceHeld) {
+ t.Fatalf("resumed under another operation's fence: proceed %v, %v", ok, err)
+ }
+ after.LiftFence(host, stored)
+ if f.gens[host] != other {
+ t.Fatalf("the resumed rollback lifted another operation's fence: fences %v", f.gens)
+ }
+ if next, err := after.NextFenceGen(); err != nil || next <= other {
+ t.Fatalf("a generation minted after the restart: %d (%v); before %d, the layout pass's %d", next, err, stored, other)
+ }
+ // Once the other operation lets go, the rollback fences at its own
+ // stored generation and proceeds, as on a cold host.
+ f.LiftFence(host, other)
+ if ok, err := after.FenceForRollback(ctx, host, stored); !ok || f.gens[host] != stored {
+ t.Fatalf("after the other fence lifted: proceed %v, %v, fences %v", ok, err, f.gens)
+ }
+}
+
+// The rollback's six hooks are all or none (provision skips a rollback
+// missing any), so production wiring must set every one; with claim control
+// on, seeds are refused until #250 and none of the swap's or the
+// rollback's hooks is armed.
+func TestWireSeedPlacementArmsTheSwapAndTheRollback(t *testing.T) {
+ ctx := context.Background()
+ data, deploy := release.NewDirStore(t.TempDir()), release.NewDirStore(t.TempDir())
+ f := &fakeFence{gens: map[string]uint64{}}
+ reg := restartTestRegistry(t)
+ nestedErr := errors.New("probe")
+ box := seedBox{Fence: f, Prefix: "instances", Swept: func() bool { return true }, NextFenceGen: fenceGenerations(reg)}
+
+ rollbackHooks := func(h *provision.SeedHooks) map[string]bool {
+ return map[string]bool{
+ "FenceForRollback": h.FenceForRollback != nil, "LiftFence": h.LiftFence != nil,
+ "RemoveLocal": h.RemoveLocal != nil, "DeleteReplica": h.DeleteReplica != nil,
+ "NextFenceGen": h.NextFenceGen != nil, "LocalSeedID": h.LocalSeedID != nil,
+ }
+ }
+
+ t.Run("armed", func(t *testing.T) {
+ resetArmedSeedHooks(t)
+ eb := &activator.ExecBackend{NestedOK: func() error { return nestedErr }}
+ wireSeedPlacement(eb, false, storeOpener(data), storeOpener(deploy), box)
+ h := armedSeedHooks.Load()
+ if h == nil || h.Data != release.Store(data) {
+ t.Fatalf("armed hooks = %+v", h)
+ }
+ for name, set := range rollbackHooks(h) {
+ if !set {
+ t.Errorf("the rollback hook %s is not wired: no rollback runs", name)
+ }
+ }
+ if h.NestedOK == nil || h.NestedOK() != nestedErr {
+ t.Error("NestedOK is not the exec backend's probe")
+ }
+ if h.LegacyOwners == nil {
+ t.Fatal("LegacyOwners is not wired")
+ }
+ body := `{"v":1,"names":{"old.example.com":{"account":"bbw","app":"shop"}}}`
+ if _, err := deploy.Put(ctx, seeds.LegacyOwnersKey, []byte(body), ""); err != nil {
+ t.Fatal(err)
+ }
+ if o, err := h.LegacyOwners(ctx); err != nil || o == nil || o.Names["old.example.com"].App != "shop" {
+ t.Errorf("LegacyOwners read %+v, %v from the record store", o, err)
+ }
+ if g, err := h.NextFenceGen(); err != nil || g == 0 {
+ t.Errorf("NextFenceGen = %d, %v", g, err)
+ }
+ f.gens["shop-1.example.com"] = 41
+ h.LiftFence("shop-1.example.com", 41)
+ if f.gens["shop-1.example.com"] != 0 {
+ t.Error("LiftFence did not reach the activator")
+ }
+ })
+ t.Run("claim control on", func(t *testing.T) {
+ resetArmedSeedHooks(t)
+ eb := &activator.ExecBackend{NestedOK: func() error { return nil }}
+ wireSeedPlacement(eb, true, storeOpener(data), storeOpener(deploy), box)
+ h := armedSeedHooks.Load()
+ if h == nil || h.Data != release.Store(data) {
+ t.Fatalf("armed hooks = %+v, want the data bucket", h)
+ }
+ for name, set := range rollbackHooks(h) {
+ if set {
+ t.Errorf("the rollback hook %s is armed with claim control on", name)
+ }
+ }
+ if h.NestedOK != nil || h.LegacyOwners != nil {
+ t.Error("the swap's hooks are armed with claim control on")
+ }
+ })
+ t.Run("no activator fence", func(t *testing.T) {
+ resetArmedSeedHooks(t)
+ wireSeedPlacement(&activator.ExecBackend{}, false, storeOpener(data), storeOpener(deploy), seedBox{Prefix: "instances"})
+ h := armedSeedHooks.Load()
+ if h == nil {
+ t.Fatal("no hooks")
+ }
+ for name, set := range rollbackHooks(h) {
+ if set {
+ t.Errorf("the rollback hook %s is armed without a fence: half the rollback", name)
+ }
+ }
+ })
+}
+
+// The rollback's step 6 deletes the host's replica and nothing beside it:
+// not a host whose name only begins with this one, not another host, not
+// the seeds/ prefix.
+func TestDeleteReplicaDeletesOnlyTheHostsReplica(t *testing.T) {
+ ctx := context.Background()
+ data := release.NewDirStore(t.TempDir())
+ const host = "shop-1.example.com"
+ put := func(k string) {
+ if _, err := data.Put(ctx, k, []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ mine := []string{"instances/" + host + "/generations/a/wal/1.ltx", "instances/" + host + "/lock.json"}
+ theirs := []string{"instances/" + host + ".au/lock.json", "instances/other.example.com/lock.json", "seeds/" + host + "/sd_aaaaaaaaaaaaaaaaaaaaaaaaaa/snapshot.db"}
+ for _, k := range append(append([]string{}, mine...), theirs...) {
+ put(k)
+ }
+ del := deleteReplica(data, "instances")
+ for _, bad := range []string{"", "..", "a/b", "."} {
+ if err := del(ctx, bad); err == nil {
+ t.Fatalf("host %q accepted", bad)
+ }
+ }
+ if err := del(ctx, host); err != nil {
+ t.Fatal(err)
+ }
+ for _, k := range mine {
+ if snapshotPresent(t, data, k) {
+ t.Errorf("%s survived", k)
+ }
+ }
+ for _, k := range theirs {
+ if !snapshotPresent(t, data, k) {
+ t.Errorf("%s was deleted", k)
+ }
+ }
+ if err := del(ctx, host); err != nil {
+ t.Fatalf("nothing there: %v", err)
+ }
+}
diff --git a/internal/provision/seedreconcile.go b/internal/provision/seedreconcile.go
index 3e87583e7..deef939cc 100644
--- a/internal/provision/seedreconcile.go
+++ b/internal/provision/seedreconcile.go
@@ -46,8 +46,11 @@ type SeedHooks struct {
// DeleteReplica deletes host's instances/ replica in the region data
// bucket. Nothing there is not an error.
DeleteReplica func(ctx context.Context, host string) error
- // NextFenceGen mints a fence generation no fence on this box holds.
- NextFenceGen func() uint64
+ // NextFenceGen mints a fence generation no fence on this box holds,
+ // never one it minted before, across agent restarts: the journal
+ // stores it, and a resumed rollback recognises and lifts its fence by
+ // it. An error, or 0, closes nothing this pass.
+ NextFenceGen func() (uint64, error)
// LocalSeedID is the seed id the <db>.seed marker beside dbPath names
// ("" for none or no marker), read through instancefs. An error is a
// marker that could not be read.
diff --git a/internal/provision/seedrollback.go b/internal/provision/seedrollback.go
index fa376d87d..c52dd33e6 100644
--- a/internal/provision/seedrollback.go
+++ b/internal/provision/seedrollback.go
@@ -95,7 +95,15 @@ func rollBackRoute(ctx context.Context, s release.Store, reg *registry.Registry,
// Step 1. The generation an earlier close stored is the one the fence
// is known by, so the journal is read back rather than trusted.
if j.State == registry.JournalSwapped {
- if err := reg.CloseImportGate(j.Host, j.SeedID, h.NextFenceGen()); err != nil {
+ gen, err := h.NextFenceGen()
+ if err == nil && gen == 0 {
+ err = errors.New("the generation source answered 0")
+ }
+ if err != nil {
+ log.Printf("provision: %s: import %s rollback deferred: no fence generation: %v", j.Host, j.SeedID, err)
+ return j, false
+ }
+ if err := reg.CloseImportGate(j.Host, j.SeedID, gen); err != nil {
log.Printf("provision: %s: import %s: close the import gate: %v", j.Host, j.SeedID, err)
return j, false
}
diff --git a/internal/provision/seedrollback_test.go b/internal/provision/seedrollback_test.go
index 0f61fcfee..7ac0f5ae2 100644
--- a/internal/provision/seedrollback_test.go
+++ b/internal/provision/seedrollback_test.go
@@ -26,7 +26,7 @@ type rollbackRecorder struct {
}
func (r *rollbackRecorder) hooks(h *SeedHooks) {
- h.NextFenceGen = func() uint64 { r.gen++; return r.gen }
+ h.NextFenceGen = func() (uint64, error) { r.gen++; return r.gen, nil }
h.FenceForRollback = func(_ context.Context, host string, gen uint64) (bool, error) {
r.steps = append(r.steps, "fence")
return r.proceed, r.fenceErr
@@ -327,10 +327,10 @@ func (b *rollbackBox) wantRestored(what string) {
}
func (b *rollbackBox) hooks(h *SeedHooks) {
- h.NextFenceGen = func() uint64 {
+ h.NextFenceGen = func() (uint64, error) {
b.steps = append(b.steps, "gen")
b.gen++
- return b.gen
+ return b.gen, nil
}
h.FenceForRollback = func(_ context.Context, host string, gen uint64) (bool, error) {
b.steps = append(b.steps, "fence")
@@ -1133,3 +1133,30 @@ func TestRollbackOfAMissingRowLeavesTheDataWhenARowAppears(t *testing.T) {
t.Fatalf("journal = %q", st)
}
}
+
+// A generation that cannot be minted (the box's source failed, or answered
+// 0, which is no fence's) closes nothing: the journal stays swapped with
+// no generation stored, so a later pass mints a real one. A 0 stored there
+// would read as "ours already" on any unfenced host.
+func TestRollbackWaitsForAFenceGeneration(t *testing.T) {
+ for name, mint := range map[string]func() (uint64, error){
+ "an error": func() (uint64, error) { return 0, errors.New("registry busy") },
+ "zero": func() (uint64, error) { return 0, nil },
+ } {
+ t.Run(name, func(t *testing.T) {
+ x := abortedImport(t)
+ x.cfg.Seeds.NextFenceGen = mint
+ x.pass(t)
+ j, _ := x.reg.ImportJournal(x.rec.Host)
+ if j == nil || j.State != registry.JournalSwapped || j.FenceGen != 0 {
+ t.Fatalf("journal = %+v, want swapped with no generation", j)
+ }
+ if got := mustGet(t, x.reg, x.rec.Host); got.ImportGate == registry.ImportGateClosed {
+ t.Fatalf("the gate closed without a generation: %+v", got)
+ }
+ if len(x.box.steps) != 0 {
+ t.Fatalf("steps = %q", x.box.steps)
+ }
+ })
+ }
+}
diff --git a/internal/registry/fencegen.go b/internal/registry/fencegen.go
new file mode 100644
index 000000000..e425cefc7
--- /dev/null
+++ b/internal/registry/fencegen.go
@@ -0,0 +1,52 @@
+package registry
+
+import (
+ "database/sql"
+ "errors"
+ "time"
+)
+
+// The box's fence generation source. The activator's fences live in the
+// agent's memory and die with it, but a generation outlives the process:
+// the import journal stores the one its rollback fences under
+// (CloseImportGate), and a resumed rollback recognises and lifts its fence
+// by that number. A generation handed out after a restart must therefore
+// never equal one handed out before it, or the resumed rollback could take
+// another operation's fence for its own, or lift it.
+//
+// Each generation is the later of the wall clock in nanoseconds (the space
+// the layout pass and the moves cut draw theirs from) and one past the
+// last generation this registry handed out, which is stored before it is
+// returned. The stored floor survives a restart, so a clock that went back
+// cannot repeat a number; the clock covers a registry restored from an
+// older snapshot.
+
+func (r *Registry) migrateFenceGenerations() error {
+ _, err := r.sql.Exec(`CREATE TABLE IF NOT EXISTS fence_generation (
+ id INTEGER PRIMARY KEY CHECK (id = 1),
+ last INTEGER NOT NULL
+);`)
+ return err
+}
+
+// NextFenceGeneration hands out a fence generation no earlier call on this
+// registry file returned, across restarts: never 0, and above every
+// generation it handed out before. now is the wall clock.
+func (r *Registry) NextFenceGeneration(now time.Time) (uint64, error) {
+ var next int64
+ err := r.inTx(func(tx txer) error {
+ var last int64
+ err := tx.QueryRow(`SELECT last FROM fence_generation WHERE id = 1`).Scan(&last)
+ if err != nil && !errors.Is(err, sql.ErrNoRows) {
+ return err
+ }
+ next = max(now.UnixNano(), last+1, 1)
+ _, err = tx.Exec(`INSERT INTO fence_generation (id, last) VALUES (1, ?)
+ON CONFLICT(id) DO UPDATE SET last = excluded.last`, next)
+ return err
+ })
+ if err != nil {
+ return 0, err
+ }
+ return uint64(next), nil
+}
diff --git a/internal/registry/fencegen_test.go b/internal/registry/fencegen_test.go
new file mode 100644
index 000000000..69fa1d9c2
--- /dev/null
+++ b/internal/registry/fencegen_test.go
@@ -0,0 +1,90 @@
+package registry
+
+import (
+ "path/filepath"
+ "sync"
+ "testing"
+ "time"
+)
+
+// A generation is never handed out twice by one registry file, across a
+// close and reopen (an agent restart) and whatever the clock says: a
+// rollback resumed after a restart compares and lifts by the generation it
+// stored before it, so a later fence must never carry the same number.
+func TestNextFenceGenerationIsUniqueAcrossRestarts(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "registry.db")
+ reg, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ now := time.Unix(1_800_000_000, 0)
+ before := map[uint64]bool{}
+ var last uint64
+ for i := 0; i < 5; i++ {
+ g, err := reg.NextFenceGeneration(now) // the same instant: still distinct
+ if err != nil {
+ t.Fatal(err)
+ }
+ if g == 0 || g <= last {
+ t.Fatalf("generation %d after %d: want nonzero and rising", g, last)
+ }
+ before[g], last = true, g
+ }
+ if err := reg.Close(); err != nil {
+ t.Fatal(err)
+ }
+
+ reg, err = Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer reg.Close()
+ // The clock went back across the restart (a box that booted before NTP
+ // synced): the generations still rise past every one handed out before.
+ for _, at := range []time.Time{now.Add(-time.Hour), now, time.Unix(0, 0), now.Add(-time.Nanosecond)} {
+ g, err := reg.NextFenceGeneration(at)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if before[g] || g <= last {
+ t.Fatalf("after the restart, at %v: generation %d (last before %d)", at, g, last)
+ }
+ last = g
+ }
+ // A clock ahead of every earlier generation is followed: generations
+ // stay in wall-clock nanoseconds, the space the layout pass and the
+ // moves cut draw theirs from.
+ later := now.Add(time.Hour)
+ if g, err := reg.NextFenceGeneration(later); err != nil || g != uint64(later.UnixNano()) {
+ t.Fatalf("a clock ahead: generation %d, %v; want %d", g, err, later.UnixNano())
+ }
+}
+
+func TestNextFenceGenerationConcurrentCallersNeverShare(t *testing.T) {
+ reg := openTestRegistry(t)
+ now := time.Unix(1_800_000_000, 0)
+ const n = 40
+ gens := make(chan uint64, n)
+ var wg sync.WaitGroup
+ for i := 0; i < n; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ g, err := reg.NextFenceGeneration(now)
+ if err != nil {
+ t.Error(err)
+ return
+ }
+ gens <- g
+ }()
+ }
+ wg.Wait()
+ close(gens)
+ seen := map[uint64]bool{}
+ for g := range gens {
+ if seen[g] {
+ t.Fatalf("generation %d handed out twice", g)
+ }
+ seen[g] = true
+ }
+}
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index e1628cf64..981509218 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -623,6 +623,9 @@ CREATE INDEX IF NOT EXISTS timers_due ON timers (next_at) WHERE next_at > 0;`);
if err := r.migrateImportJournal(); err != nil {
return err
}
+ if err := r.migrateFenceGenerations(); err != nil {
+ return err
+ }
if err := r.migrateAccounts(); err != nil {
return err
}
diff --git a/internal/seeds/floor.go b/internal/seeds/floor.go
index 0f0ad76df..00158620b 100644
--- a/internal/seeds/floor.go
+++ b/internal/seeds/floor.go
@@ -11,9 +11,17 @@ import (
"github.com/carlosframework/platform/internal/release"
)
-// ProtocolVersion is the seeds protocol this build speaks. 0 until the box
-// can place, swap and roll back a seed (Plan 7 raises it to 1): a box that
-// reports 0 can never satisfy the console's RequiredProtocol.
+// ProtocolVersion is the seeds protocol this build speaks. A box that
+// reports 0 can never satisfy the console's RequiredProtocol, so every
+// seed door stays shut.
+//
+// It stays 0 through Plan 7, although this build places, swaps and rolls
+// back a seed (Plans 4 and 7): spec §5.8's move refusals, the seeds sweep
+// and the legacy-routes assign door must exist before any door opens
+// (preflight M9). Plan 8's last task raises it to 1, with the comment
+// "1: this build places, swaps and rolls back a seed", and rewrites
+// floor_test.go's TestProtocolStartsBelowWhatTheConsoleRequires as
+// TestProtocolMeetsWhatTheConsoleRequires.
const ProtocolVersion = 0
// RequiredProtocol is what the console requires every box in the region to
diff --git a/internal/seeds/floor_test.go b/internal/seeds/floor_test.go
index 373ce0889..111be296c 100644
--- a/internal/seeds/floor_test.go
+++ b/internal/seeds/floor_test.go
@@ -109,9 +109,12 @@ func TestReleaseAtOrAbove(t *testing.T) {
}
func TestProtocolStartsBelowWhatTheConsoleRequires(t *testing.T) {
- // Plan 3 ships the floor dark: no box built from it can satisfy the
- // console's check. Plan 7 Task 7 raises ProtocolVersion and rewrites
- // this test.
+ // The floor ships dark: no box built from it can satisfy the console's
+ // check. Plan 7 builds the swap and the rollback but leaves the switch
+ // off (preflight M9): spec §5.8's move refusals, the seeds sweep and
+ // the legacy-routes assign door must exist before any door opens. Plan
+ // 8's last task raises ProtocolVersion and rewrites this test as
+ // TestProtocolMeetsWhatTheConsoleRequires.
if ProtocolVersion >= RequiredProtocol {
t.Fatalf("ProtocolVersion = %d, RequiredProtocol = %d", ProtocolVersion, RequiredProtocol)
}
diff --git a/internal/seeds/seeds_test.go b/internal/seeds/seeds_test.go
index c33836e54..108f3693a 100644
--- a/internal/seeds/seeds_test.go
+++ b/internal/seeds/seeds_test.go
@@ -4,6 +4,8 @@ import (
"errors"
"strings"
"testing"
+
+ "github.com/carlosframework/platform/internal/registry"
)
func TestNewIDShape(t *testing.T) {
@@ -169,3 +171,12 @@ func TestRowSeedEncoding(t *testing.T) {
}
}
}
+
+// The edge refuses a row whose import gate is the registry's closed value,
+// and the console and the box name it as seeds.ClosedGate: the two must
+// be one value (seeds may import registry; registry does not import seeds).
+func TestClosedGateIsTheRegistrysClosedGate(t *testing.T) {
+ if ClosedGate != registry.ImportGateClosed {
+ t.Fatalf("seeds.ClosedGate = %q, registry.ImportGateClosed = %q", ClosedGate, registry.ImportGateClosed)
+ }
+}