amadan

carlos / platform Public

host: relocation refuses what it cannot read: an unreadable completed relocation holds the barrier and the sweep, an empty listing needs session 0 and the replicator's own zero position, and a malformed adoption record fails the flush

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude 95a1ed845ef24325324319bf30869b7932f5e636 parent 85e58f6
20 files changed, +1013 −161
  • internal/activator/fenceowner_test.go +62 −0
  • internal/activator/recoveryprobe.go +3 −0
  • internal/activator/recoveryprobe_sweep_test.go +11 −0
  • internal/host/relocationassembly.go +61 −19
  • internal/host/relocationassembly_test.go +259 −0
  • internal/host/relocationeffects.go +0 −29
  • internal/host/relocationflush.go +35 −37
  • internal/host/relocationflush_adoption_test.go +47 −8
  • internal/host/relocationsweep.go +133 −45
  • internal/host/relocationsweep_test.go +119 −0
  • internal/hostclaim/relocation.go +6 −4
  • internal/hostclaim/relocation_probestop.go +33 −9
  • internal/hostclaim/relocation_probestop_test.go +49 −4
  • internal/hostclaim/relocation_test.go +1 −1
  • internal/hostrelocate/participant_gate_test.go +49 −0
  • internal/hostrelocate/participant_test.go +2 −1
  • internal/hostrelocate/worker.go +11 −4
  • internal/hostrelocate/worker_bounds_test.go +54 −0
  • internal/lease/lease.go +15 −0
  • internal/lease/position_test.go +63 −0
diff --git a/internal/activator/fenceowner_test.go b/internal/activator/fenceowner_test.go
index 540b8f3ea..25c1054a5 100644
--- a/internal/activator/fenceowner_test.go
+++ b/internal/activator/fenceowner_test.go
@@ -110,3 +110,65 @@ func TestLiftFenceAsLiftsOnlyTheOwnersFence(t *testing.T) {
t.Fatal("a plain lift left the owner behind")
}
}
+
+// TestAnOwnedCallNeverTakesBackAFenceReplacedWhileItWaited (fix round 1,
+// M1; ruling 75): FenceAndStopAs installs its owned fence and waits for
+// the stop. Meanwhile a plain FenceAndStop installs at the same generation,
+// or a plain LiftFence lifts it. When the stop completes the owned call
+// answers ErrStopUnproven and leaves the fence as the other caller left
+// it: still plain and unowned, or still lifted, never the owner's again.
+func TestAnOwnedCallNeverTakesBackAFenceReplacedWhileItWaited(t *testing.T) {
+ for _, meanwhile := range []string{"a plain fence", "a plain lift"} {
+ t.Run(meanwhile, func(t *testing.T) {
+ a, e, _, _, _ := maintenanceAwake(t)
+ stopped := make(chan struct{})
+ a.mu.Lock()
+ e.state, e.stopped = actStopping, stopped
+ a.mu.Unlock()
+ type answer struct {
+ proof *StopEvidence
+ err error
+ }
+ done := make(chan answer, 1)
+ go func() {
+ proof, err := a.FenceAndStopAs(maintenanceContext(t, 5*time.Second), e.host, 12, "owner-1")
+ done <- answer{proof, err}
+ }()
+ for installed := false; !installed; {
+ a.mu.Lock()
+ installed = a.fences[e.host] == 12 && a.fenceOwners[e.host] == "owner-1"
+ a.mu.Unlock()
+ time.Sleep(time.Millisecond)
+ }
+ switch meanwhile {
+ case "a plain fence":
+ if _, err := a.FenceAndStop(maintenanceContext(t, 20*time.Millisecond), e.host, 12); !errors.Is(err, ErrStopUnproven) {
+ t.Fatalf("the plain fence: %v", err)
+ }
+ case "a plain lift":
+ if !a.LiftFence(e.host, 12) {
+ t.Fatal("the plain lift lifted nothing")
+ }
+ }
+ a.mu.Lock()
+ delete(a.entries, e.host)
+ a.mu.Unlock()
+ close(stopped)
+ got := <-done
+ if !errors.Is(got.err, ErrStopUnproven) || got.proof != nil {
+ t.Fatalf("the owned call after its fence was replaced: %v %v", got.proof, got.err)
+ }
+ a.mu.Lock()
+ gen := a.fences[e.host]
+ owner, owned := a.fenceOwners[e.host]
+ a.mu.Unlock()
+ want := uint64(12)
+ if meanwhile == "a plain lift" {
+ want = 0
+ }
+ if gen != want || owned {
+ t.Fatalf("the fence is %d owned by %q (%v), want %d unowned", gen, owner, owned, want)
+ }
+ })
+ }
+}
diff --git a/internal/activator/recoveryprobe.go b/internal/activator/recoveryprobe.go
index 3783923bf..f82a54e9d 100644
--- a/internal/activator/recoveryprobe.go
+++ b/internal/activator/recoveryprobe.go
@@ -321,8 +321,11 @@ func (a *Service) reviewRecoveryHold(e *actEntry) {
// The entry retains both the observed fence and the adapter's
// owning generation. Check and remove under the same lock as the
// hold transition; never lift a replacement operation's fence.
+ // A lift clears an owned fence's owner with it (FenceAndStopAs),
+ // so no owner outlives its fence.
if e.recoveryPolicy.RecoveryFence != 0 && a.fences[e.host] == e.recoveryPolicy.RecoveryFence {
delete(a.fences, e.host)
+ delete(a.fenceOwners, e.host)
}
log.Printf("activator: %s source opened on the recovery probe's session; serving", e.host)
if e.gen.N != 0 {
diff --git a/internal/activator/recoveryprobe_sweep_test.go b/internal/activator/recoveryprobe_sweep_test.go
index 8f03fe3ae..fa83dfd95 100644
--- a/internal/activator/recoveryprobe_sweep_test.go
+++ b/internal/activator/recoveryprobe_sweep_test.go
@@ -89,6 +89,11 @@ func TestRecoveryProbeSweepOpensOnlyExactSessionAndFence(t *testing.T) {
for _, phase := range []string{hostclaim.PhaseProbing, hostclaim.PhaseProbed, hostclaim.PhaseOpened, "serving", "new-session", "foreign-executor", "unreadable"} {
t.Run(phase, func(t *testing.T) {
a, r, e, proc, _, admitted := sweptProbe(t)
+ // The cut fence carries an owner (FenceAndStopAs at the same
+ // generation): the opening's lift clears it with the fence.
+ a.mu.Lock()
+ a.fenceOwners = map[string]string{probeHost: "owner-1"}
+ a.mu.Unlock()
fresh := admitted
c, h := *admitted.Claim, *admitted.Claim.Holder
c.Holder = &h
@@ -128,6 +133,12 @@ func TestRecoveryProbeSweepOpensOnlyExactSessionAndFence(t *testing.T) {
if a.Fenced(probeHost) || proc.isStopped() {
t.Fatal("exact opening did not release the cut fence and retain its process")
}
+ a.mu.Lock()
+ _, owned := a.fenceOwners[probeHost]
+ a.mu.Unlock()
+ if owned {
+ t.Fatal("the opening lifted the cut fence but left its owner")
+ }
release, err := a.beginRequest(t.Context(), registry.Route{Host: probeHost})
if err != nil {
t.Fatalf("opened session cannot serve: %v", err)
diff --git a/internal/host/relocationassembly.go b/internal/host/relocationassembly.go
index 9fcbd0708..a16b9b91a 100644
--- a/internal/host/relocationassembly.go
+++ b/internal/host/relocationassembly.go
@@ -25,8 +25,12 @@ import (
// RelocationActivator is the part of activator.Service relocation uses:
// the probe attempt (Task 8d) and the local fence that ends a passed
// probe's session before the serving CAS (ruling 49). The fence is an
-// owned one (ruling 75): taken and met again only under the owner the
-// probe stop persisted, and lifted only under it.
+// owned one (ruling 75): relocation takes and meets it again only under
+// the owner the probe stop persisted, and lifts it only under that owner.
+// The activator's other paths do not ask for the owner: a plain
+// FenceAndStop or LiftFence at the same generation, or a recovery probe
+// opening on a RecoveryFence equal to it, replaces or lifts it and clears
+// the owner, after which the owner can neither meet nor lift it.
type RelocationActivator interface {
ProbeWake(context.Context, activator.ProbeConfig) (activator.ProbeResult, error)
FenceAndStopAs(context.Context, string, uint64, string) (*activator.StopEvidence, error)
@@ -144,8 +148,9 @@ func relocationPathsFor(dataDir string, rt registry.Route) relocationPaths {
}
// relocationHostPaths reads the host's own row: the source's flush
-// publishes the paths its writer used. No row is no paths, which the
-// flush refuses rather than guess.
+// publishes the paths its writer used. No row, or a row that cannot be
+// read, is no paths, and ReleaseFlushFiles.Flush refuses empty paths
+// before it stages anything rather than guess.
func relocationHostPaths(dataDir string, reg *registry.Registry) func(string) relocationPaths {
return func(host string) relocationPaths {
if reg == nil {
@@ -159,10 +164,8 @@ func relocationHostPaths(dataDir string, reg *registry.Registry) func(string) re
}
}
-// relocationFlushDB is release step 2 (ruling 8): SyncAndWait with every
-// writer stopped, then N is the newest level-0 MaxTXID, which must equal
-// the plan's MaxTXID. A listing with no LTX file is the stopped writer
-// vouching that it never committed: N is 0.
+// relocationFlushDB is release step 2 (ruling 8) for the session's own
+// replicator: releaseFinalTXID over the claim wake's attempt for the host.
func (a ClaimAssembly) relocationFlushDB(cfg RelocationAssemblyConfig) func(context.Context, hostclaim.Observed) (uint64, error) {
return func(ctx context.Context, o hostclaim.Observed) (uint64, error) {
if a.Wake == nil || o.Claim == nil || o.Claim.Holder == nil {
@@ -172,19 +175,58 @@ func (a ClaimAssembly) relocationFlushDB(cfg RelocationAssemblyConfig) func(cont
if att == nil || att.rep == nil || cfg.Versions == nil || cfg.Bucket == "" {
return 0, ErrClaimStopUnavailable
}
- if err := att.rep.SyncAndWait(ctx); err != nil {
- return 0, err
+ sel := func(ctx context.Context, r genselect.Request) (*genselect.Plan, error) {
+ return genselect.Select(ctx, cfg.Versions, r)
}
- plan, err := genselect.Select(ctx, cfg.Versions, genselect.Request{Bucket: cfg.Bucket, Region: cfg.Region, Host: o.Host, Holder: *o.Claim.Holder})
- if errors.Is(err, genselect.ErrEmpty) {
+ return releaseFinalTXID(ctx, att.rep, sel, genselect.Request{Bucket: cfg.Bucket, Region: cfg.Region, Host: o.Host, Holder: *o.Claim.Holder})
+ }
+}
+
+// releaseReplica is the part of the session's replicator the release's
+// database flush uses (lease.Replicator).
+type releaseReplica interface {
+ SyncAndWait(context.Context) error
+ Position() (uint64, error)
+}
+
+// releaseFinalTXID is release step 2 (ruling 8): SyncAndWait with every
+// writer stopped, then N is the newest level-0 MaxTXID, which must equal
+// the plan's MaxTXID.
+//
+// An empty listing is not evidence of an empty database (fix round 1, I2):
+// a wrong bucket, or a listing not yet visible, reads the same. N is 0
+// only when the holder's writer session is 0 (no writer was ever admitted,
+// pin.go's "session-0" and ruling 25) and the replicator's own position
+// (lease.Replicator.Position, litestream's newest local LTX) is 0 too.
+// Any other empty listing refuses: the release is never journaled, and the
+// worker marks the op stuck ReleaseDeadline after releasing began, for an
+// operator, as a refused pin leaves it.
+func releaseFinalTXID(ctx context.Context, rep releaseReplica, sel func(context.Context, genselect.Request) (*genselect.Plan, error), req genselect.Request) (uint64, error) {
+ if rep == nil || sel == nil {
+ return 0, ErrClaimStopUnavailable
+ }
+ if err := rep.SyncAndWait(ctx); err != nil {
+ return 0, err
+ }
+ pos, err := rep.Position()
+ if err != nil {
+ return 0, err
+ }
+ plan, err := sel(ctx, req)
+ if errors.Is(err, genselect.ErrEmpty) {
+ if req.Holder.Session == 0 && pos == 0 {
return 0, nil
}
- if err != nil {
- return 0, err
- }
- if plan.MaxLevel0TXID != plan.MaxTXID {
- return 0, fmt.Errorf("%w: level 0 at %d, the database at %d", ErrClaimStopUnavailable, plan.MaxLevel0TXID, plan.MaxTXID)
- }
- return plan.MaxTXID, nil
+ return 0, fmt.Errorf("%w: the listing is empty for writer session %d, whose replicator is at %d", ErrClaimStopUnavailable, req.Holder.Session, pos)
+ }
+ if err != nil {
+ return 0, err
+ }
+ if plan == nil {
+ return 0, ErrClaimStopUnavailable
+ }
+ if plan.MaxLevel0TXID != plan.MaxTXID {
+ return 0, fmt.Errorf("%w: level 0 at %d, the database at %d", ErrClaimStopUnavailable, plan.MaxLevel0TXID, plan.MaxTXID)
}
+ return plan.MaxTXID, nil
}
diff --git a/internal/host/relocationassembly_test.go b/internal/host/relocationassembly_test.go
new file mode 100644
index 000000000..7d5a9bb40
--- /dev/null
+++ b/internal/host/relocationassembly_test.go
@@ -0,0 +1,259 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "slices"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ execctl "github.com/carlosframework/platform/internal/hostexec"
+ "github.com/carlosframework/platform/internal/hostrelocate"
+)
+
+// fakeReleaseReplica is the session's replicator as releaseFinalTXID sees
+// it: SyncAndWait, then its own position.
+type fakeReleaseReplica struct {
+ calls []string
+ syncErr error
+ pos uint64
+ posErr error
+}
+
+func (r *fakeReleaseReplica) SyncAndWait(context.Context) error {
+ r.calls = append(r.calls, "sync")
+ return r.syncErr
+}
+
+func (r *fakeReleaseReplica) Position() (uint64, error) {
+ r.calls = append(r.calls, "position")
+ return r.pos, r.posErr
+}
+
+// Fix round 1, I2: an empty listing is not evidence of an empty database.
+// FinalTXID 0 comes only from a writer session 0 whose replicator is at 0
+// too; a wrong bucket (an empty listing while the replicator reports a
+// position) refuses, and so does an empty listing for an admitted writer.
+// A listing is read only after the replicator synced and reported.
+func TestReleaseFinalTXIDNeedsTheReplicatorsOwnEvidence(t *testing.T) {
+ ctx := context.Background()
+ failure := errors.New("s3: access denied")
+ for _, tc := range []struct {
+ name string
+ session uint64
+ rep fakeReleaseReplica
+ plan *genselect.Plan
+ selErr error
+ want uint64
+ refuses bool
+ calls []string
+ }{
+ {name: "the wrong bucket: an empty listing, the replicator at 41", session: 6, rep: fakeReleaseReplica{pos: 41}, selErr: genselect.ErrEmpty, refuses: true},
+ {name: "an empty listing for an admitted writer whose replicator is at 0", session: 6, selErr: genselect.ErrEmpty, refuses: true},
+ {name: "an empty listing for session 0 whose replicator is at 3", rep: fakeReleaseReplica{pos: 3}, selErr: genselect.ErrEmpty, refuses: true},
+ {name: "a genuinely empty session-0 database", selErr: genselect.ErrEmpty, want: 0},
+ {name: "a listing at the database's position", session: 6, rep: fakeReleaseReplica{pos: 41}, plan: &genselect.Plan{MaxTXID: 41, MaxLevel0TXID: 41}, want: 41},
+ {name: "level 0 behind the database", session: 6, rep: fakeReleaseReplica{pos: 41}, plan: &genselect.Plan{MaxTXID: 41, MaxLevel0TXID: 40}, refuses: true},
+ {name: "a listing that fails", session: 6, rep: fakeReleaseReplica{pos: 41}, selErr: failure, refuses: true},
+ {name: "a sync that fails", session: 6, rep: fakeReleaseReplica{syncErr: failure}, refuses: true, calls: []string{"sync"}},
+ {name: "a position that cannot be read", session: 0, rep: fakeReleaseReplica{posErr: failure}, refuses: true, calls: []string{"sync", "position"}},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ rep := tc.rep
+ selected := false
+ sel := func(_ context.Context, r genselect.Request) (*genselect.Plan, error) {
+ selected = true
+ if r.Holder.Session != tc.session || r.Host != "shop.example" {
+ t.Fatalf("selected for %+v", r)
+ }
+ return tc.plan, tc.selErr
+ }
+ got, err := releaseFinalTXID(ctx, &rep, sel, genselect.Request{Host: "shop.example", Holder: hostclaim.Holder{Session: tc.session}})
+ if tc.refuses {
+ if err == nil {
+ t.Fatalf("released at %d", got)
+ }
+ } else if err != nil || got != tc.want {
+ t.Fatalf("got %d %v, want %d", got, err, tc.want)
+ }
+ want := tc.calls
+ if want == nil {
+ want = []string{"sync", "position"}
+ }
+ if !slices.Equal(rep.calls, want) || selected != (tc.calls == nil) {
+ t.Fatalf("ran %v selected=%v, want %v", rep.calls, selected, want)
+ }
+ })
+ }
+}
+
+// relocationFlushDB has nothing to flush without the session's replicator:
+// no wake, no attempt for the host, no data bucket.
+func TestRelocationFlushDBNeedsTheSessionsReplicator(t *testing.T) {
+ obs := hostclaim.Observed{Host: "shop.example", Claim: &hostclaim.Claim{Host: "shop.example", Holder: &hostclaim.Holder{Session: 0}}}
+ cfg := RelocationAssemblyConfig{Bucket: "data"}
+ for name, a := range map[string]ClaimAssembly{"no wake": {}, "no attempt": {Wake: &ClaimWakeAssembly{}}} {
+ if _, err := a.relocationFlushDB(cfg)(context.Background(), obs); !errors.Is(err, ErrClaimStopUnavailable) {
+ t.Fatalf("%s: %v", name, err)
+ }
+ }
+ if _, err := (ClaimAssembly{Wake: &ClaimWakeAssembly{}}).relocationFlushDB(cfg)(context.Background(), hostclaim.Observed{Host: "shop.example"}); !errors.Is(err, ErrClaimStopUnavailable) {
+ t.Fatalf("no claim: %v", err)
+ }
+}
+
+// relocationRoot is root's isolated launcher as relocationPreflight asks
+// it: the protocol, then the capability for one kind.
+type relocationRoot struct {
+ protocol string
+ kindErr error
+ asked []string
+}
+
+func (r *relocationRoot) Capability(context.Context) (string, error) {
+ r.asked = append(r.asked, "protocol")
+ return r.protocol, nil
+}
+func (*relocationRoot) Launch(context.Context, hostbox.RoleLaunchRequest) (hostbox.RoleLaunchGrant, error) {
+ return hostbox.RoleLaunchGrant{}, errors.New("unused")
+}
+func (*relocationRoot) Stop(context.Context, hostbox.RoleLaunchRequest) error { return nil }
+func (r *relocationRoot) KindCapability(_ context.Context, kind string) error {
+ r.asked = append(r.asked, "kind "+kind)
+ return r.kindErr
+}
+
+// Fix round 1, M4: relocationPreflight (ruling 48) asks root for the
+// isolated launch protocol, then for kind relocating; either refusal is
+// the answer. After P's opening, until claim ask C11, a later attempt is
+// ErrLaterAttemptUnlaunchable without asking root (ruling 58).
+func TestRelocationPreflightAsksRootForTheRelocatingKind(t *testing.T) {
+ ctx := context.Background()
+ refused := errors.New("root: no attempt reader for relocating")
+ probing := func(opened bool) hostclaim.Observed {
+ j := map[string]string{"prepared": "P1"}
+ if opened {
+ j["prepared_opened"] = "P1"
+ }
+ return hostclaim.Observed{Host: "shop.example", Claim: &hostclaim.Claim{Host: "shop.example", Holder: &hostclaim.Holder{Session: 7},
+ Op: hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "rel-1", Phase: hostclaim.PhaseRelocatingProbing, Journal: j}}}
+ }
+ for _, tc := range []struct {
+ name string
+ root *relocationRoot
+ opened bool
+ later bool
+ want error
+ asked []string
+ }{
+ {name: "root answers both", root: &relocationRoot{protocol: hostbox.RoleLaunchProtocol}, asked: []string{"protocol", "kind relocating"}},
+ {name: "root speaks another protocol", root: &relocationRoot{protocol: "old"}, want: activator.ErrClaimIsolationUnavailable, asked: []string{"protocol"}},
+ {name: "root refuses the kind", root: &relocationRoot{protocol: hostbox.RoleLaunchProtocol, kindErr: refused}, want: refused, asked: []string{"protocol", "kind relocating"}},
+ {name: "a later attempt before C11", root: &relocationRoot{protocol: hostbox.RoleLaunchProtocol}, opened: true, want: hostrelocate.ErrLaterAttemptUnlaunchable},
+ {name: "a later attempt once C11 is answered", root: &relocationRoot{protocol: hostbox.RoleLaunchProtocol}, opened: true, later: true, asked: []string{"protocol", "kind relocating"}},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ a := ClaimAssembly{Wake: &ClaimWakeAssembly{Roles: &ClaimRoleAssembly{Root: tc.root}}}
+ err := a.relocationPreflight(RelocationAssemblyConfig{LaterProbeAttempts: tc.later})(ctx, probing(tc.opened))
+ if (tc.want == nil && err != nil) || (tc.want != nil && !errors.Is(err, tc.want)) {
+ t.Fatalf("err = %v, want %v", err, tc.want)
+ }
+ if !slices.Equal(tc.root.asked, tc.asked) {
+ t.Fatalf("asked root %v, want %v", tc.root.asked, tc.asked)
+ }
+ })
+ }
+ if err := (ClaimAssembly{}).relocationPreflight(RelocationAssemblyConfig{})(ctx, probing(false)); !errors.Is(err, activator.ErrClaimIsolationUnavailable) {
+ t.Fatalf("no roles: %v", err)
+ }
+}
+
+// Fix round 1, M4 (D5d): Acknowledge's one binding check. Main's
+// controller must be for exactly the observed op, incarnation and host;
+// any other is ErrLost, never an acknowledgement.
+func TestRelocationAcknowledgeBindsTheObservedOperation(t *testing.T) {
+ ctx := context.Background()
+ e, obs, _, _ := effectsFixture(t)
+ ack, err := e.Acknowledge(ctx, obs)
+ if err != nil || ack.Op != "rel-1" || ack.ControllerID != "c9" || ack.LeaseETag != "lease-1" || ack.Enrollment != "enr" || ack.BootID != "boot" {
+ t.Fatalf("the bound acknowledgement: %+v %v", ack, err)
+ }
+ for name, change := range map[string]func(*execctl.State){
+ "another op": func(s *execctl.State) { s.Op = "rel-0" },
+ "another incarnation": func(s *execctl.State) { s.Incarnation = "inc-0" },
+ "another host": func(s *execctl.State) { s.Host = "other.example" },
+ } {
+ e, obs, _, _ := effectsFixture(t)
+ inner := e.Participation
+ e.Participation = func(ctx context.Context, o hostclaim.Observed) (execctl.State, error) {
+ st, err := inner(ctx, o)
+ change(&st)
+ return st, err
+ }
+ if ack, err := e.Acknowledge(ctx, obs); !errors.Is(err, execctl.ErrLost) {
+ t.Fatalf("%s: acknowledged %+v %v", name, ack, err)
+ }
+ if err := e.Ready(ctx, obs); !errors.Is(err, execctl.ErrLost) {
+ t.Fatalf("%s: ready %v", name, err)
+ }
+ }
+}
+
+// Fix round 1, M4: validOpID is the one op-id check for a name under the
+// data root. Release positions refuse every id it refuses.
+func TestValidOpIDRefusesEveryPathThatIsNotOneName(t *testing.T) {
+ for _, op := range []string{"", ".", "..", "a/b", `a\b`, "a\x00b", "../rel-1", "rel-1/.."} {
+ if validOpID(op) {
+ t.Errorf("%q accepted", op)
+ }
+ f := ReleasePositionsFile{Dir: t.TempDir()}
+ r := StopRequest{Host: "shop.example", OperationID: op}
+ if err := f.Record(context.Background(), ReleasePosition{Request: r, PID: 1, ProcessStart: "1"}); !errors.Is(err, ErrClaimStopUnavailable) {
+ t.Errorf("%q recorded: %v", op, err)
+ }
+ if _, _, err := f.Recorded(context.Background(), r); !errors.Is(err, ErrClaimStopUnavailable) {
+ t.Errorf("%q read: %v", op, err)
+ }
+ }
+ for _, op := range []string{"rel-1", ".rel", "rel..1", "a.b"} {
+ if !validOpID(op) {
+ t.Errorf("%q refused", op)
+ }
+ }
+}
+
+// Fix round 1, M4: the release's Collect provider is a recovery request
+// (StopRequest.Recover) exactly when the release offers the recovery
+// drain, after a restart; the process that flushed asks with its own
+// producers and no Recover.
+func TestRelocationCollectRecoversAfterARestart(t *testing.T) {
+ ctx := context.Background()
+ e, obs, _, _ := effectsFixture(t)
+ var asked []StopRequest
+ a := ClaimAssembly{collectStop: func(ctx context.Context, r StopRequest, d StopDrainer) (StopResult, error) {
+ asked = append(asked, r)
+ if !r.Recover {
+ if _, err := d.DrainClaim(ctx, r, "fence"); err != nil {
+ return StopResult{}, err
+ }
+ }
+ return StopResult{Key: "control/hosts/shop.example/evidence/rel-1/src/r.json", Digest: strings.Repeat("a", 64)}, nil
+ }}
+ e.Collect = a.RelocationEffects(RelocationAssemblyConfig{}, nil).Collect
+ if _, err := e.Release(ctx, obs); err != nil {
+ t.Fatal(err)
+ }
+ e.Identity = func(context.Context) (RelocationIdentity, error) {
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 4}, BootID: "boot-2", ControllerID: "ctl-2", PID: 200, ProcessStart: "9000"}, nil
+ }
+ if _, err := e.Release(ctx, obs); err != nil {
+ t.Fatal(err)
+ }
+ if len(asked) != 2 || asked[0].Recover || !asked[1].Recover || asked[1].BootID != "boot" || asked[1].ControllerID != "ctl" {
+ t.Fatalf("asked root %+v", asked)
+ }
+}
diff --git a/internal/host/relocationeffects.go b/internal/host/relocationeffects.go
index 0be97f348..960df4242 100644
--- a/internal/host/relocationeffects.go
+++ b/internal/host/relocationeffects.go
@@ -4,8 +4,6 @@ import (
"context"
"encoding/json"
"fmt"
- "os"
- "path/filepath"
"github.com/carlosframework/platform/internal/hostclaim"
execctl "github.com/carlosframework/platform/internal/hostexec"
@@ -20,33 +18,6 @@ import (
// claimLaunchOwner.local rule.
type RelocationIdentity = claimStopLocal
-// writeFileSynced writes path whole: a synced temporary file renamed into
-// place, then the directory synced, so a committed entry survives a crash.
-func writeFileSynced(path string, body []byte, mode os.FileMode) error {
- tmp := path + ".tmp"
- if err := os.WriteFile(tmp, body, mode); err != nil {
- return err
- }
- f, err := os.Open(tmp)
- if err != nil {
- return err
- }
- if err := f.Sync(); err != nil {
- _ = f.Close()
- return err
- }
- _ = f.Close()
- if err := os.Rename(tmp, path); err != nil {
- return err
- }
- dir, err := os.Open(filepath.Dir(path))
- if err != nil {
- return err
- }
- defer dir.Close()
- return dir.Sync()
-}
-
// ReleasePosition is one recorded release: the exact stop request, the
// process that flushed it (PID and start time, beside the request's boot
// and controller), and the positions its flush reached.
diff --git a/internal/host/relocationflush.go b/internal/host/relocationflush.go
index ef88ff0a1..5a73cf959 100644
--- a/internal/host/relocationflush.go
+++ b/internal/host/relocationflush.go
@@ -49,6 +49,12 @@ func (f ReleaseFlushFiles) Flush(ctx context.Context, obs hostclaim.Observed) (i
if h.Generation == "" || h.Location == "" || !controlname.ValidHostShape(obs.Host) {
return 0, ErrClaimStopUnavailable
}
+ // No paths is no flush, refused before anything is staged: the host
+ // has no route (relocationHostPaths), or its route could not be read.
+ envPath, filesDir := f.EnvPath(obs.Host), f.FilesDir(obs.Host)
+ if envPath == "" || filesDir == "" {
+ return 0, fmt.Errorf("%w: no files directory or env file for %s", ErrClaimStopUnavailable, obs.Host)
+ }
p, err := hostfiles.Open(ctx, f.Data, h.Location+"/files/"+h.Generation+"/")
if err != nil {
return 0, err
@@ -60,7 +66,7 @@ func (f ReleaseFlushFiles) Flush(ctx context.Context, obs hostclaim.Observed) (i
if err := hostfiles.StageWake(ctx, p, ws); err != nil {
return 0, err
}
- env, err := os.ReadFile(f.EnvPath(obs.Host))
+ env, err := os.ReadFile(envPath)
switch {
case err == nil:
if err := p.StagePlatform(ctx, hostfiles.EnvItem, env); err != nil {
@@ -76,22 +82,36 @@ func (f ReleaseFlushFiles) Flush(ctx context.Context, obs hostclaim.Observed) (i
return 0, err
}
// O5 (instance-key plan rev 3): the adoption record beside the env
- // file is staged in the same commit when it parses and binds exactly
- // the releasing writer; otherwise its deletion is staged, so a stale
- // record, or none, is never published as this writer's.
+ // file is staged in the same commit when it binds exactly the
+ // releasing writer. A record of another writer (an earlier session or
+ // generation), or none, has its deletion staged, so it is never
+ // published as this writer's. A record that is present but does not
+ // parse refuses the flush: it may be this writer's, and absence is
+ // never evidence (fix round 1, M9).
writer := instancekey.Binding{Host: obs.Host, Incarnation: h.Incarnation, Session: h.Session, Generation: h.Generation}
- rec, err := os.ReadFile(filepath.Join(filepath.Dir(f.EnvPath(obs.Host)), obs.Host+instancekey.AdoptionSuffix))
- if err != nil && !errors.Is(err, os.ErrNotExist) {
- return 0, err
- }
- if a, perr := instancekey.ParseAdoption(rec); err == nil && perr == nil && a.Binds(writer) {
- if err := p.StagePlatform(ctx, instancekey.AdoptionItem, rec); err != nil {
+ rec, err := os.ReadFile(filepath.Join(filepath.Dir(envPath), obs.Host+instancekey.AdoptionSuffix))
+ switch {
+ case errors.Is(err, os.ErrNotExist):
+ if err := p.StagePlatformDelete(instancekey.AdoptionItem); err != nil {
return 0, err
}
- } else if err := p.StagePlatformDelete(instancekey.AdoptionItem); err != nil {
+ case err != nil:
return 0, err
+ default:
+ a, err := instancekey.ParseAdoption(rec)
+ if err != nil {
+ return 0, fmt.Errorf("%w: release flush of %s: %w", ErrClaimStopUnavailable, obs.Host, err)
+ }
+ if a.Binds(writer) {
+ err = p.StagePlatform(ctx, instancekey.AdoptionItem, rec)
+ } else {
+ err = p.StagePlatformDelete(instancekey.AdoptionItem)
+ }
+ if err != nil {
+ return 0, err
+ }
}
- st, err := hostfiles.SyncDir(ctx, p, f.FilesDir(obs.Host), f.Skip)
+ st, err := hostfiles.SyncDir(ctx, p, filesDir, f.Skip)
if err != nil {
return 0, err
}
@@ -144,31 +164,9 @@ func (f ReleasePositionsFile) Record(_ context.Context, p ReleasePosition) error
if err != nil {
return err
}
- tmp := path + ".tmp"
- w, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o600)
- if err != nil {
- return err
- }
- if _, err := w.Write(raw); err != nil {
- _ = w.Close()
- return err
- }
- if err := w.Sync(); err != nil {
- _ = w.Close()
- return err
- }
- if err := w.Close(); err != nil {
- return err
- }
- if err := os.Rename(tmp, path); err != nil {
- return err
- }
- dir, err := os.Open(f.Dir)
- if err != nil {
- return err
- }
- defer dir.Close()
- return dir.Sync()
+ // Moves' atomic write (moveslandinstall.go): a synced temporary in
+ // the same directory, renamed into place, the directory synced.
+ return writeFileAtomic(path, raw, 0o600)
}
// sameStop is the installation and session, not the controller.
diff --git a/internal/host/relocationflush_adoption_test.go b/internal/host/relocationflush_adoption_test.go
index eda94903d..35417a975 100644
--- a/internal/host/relocationflush_adoption_test.go
+++ b/internal/host/relocationflush_adoption_test.go
@@ -2,6 +2,7 @@ package host
import (
"context"
+ "errors"
"os"
"path/filepath"
"testing"
@@ -15,20 +16,23 @@ import (
// O5 (instance-key plan rev 3): the release flush publishes the adoption
// record in the same commit as the env item only when it binds exactly the
-// releasing writer; a stale record, or none, publishes none.
+// releasing writer; a stale record, or none, publishes none. A record that
+// does not parse refuses the flush (fix round 1, M9): it may be this
+// writer's, and absence is never evidence.
func TestReleaseFlushPublishesOnlyTheReleasingWritersAdoption(t *testing.T) {
ctx := context.Background()
const host = "shop.example"
writer := instancekey.Binding{Host: host, Incarnation: "inc-1", Session: 6, Generation: "g6"}
for _, tc := range []struct {
- name string
- record []byte // nil is no file
- want bool
+ name string
+ record []byte // nil is no file
+ want bool
+ refuses bool
}{
- {"the releasing writer's record", instancekey.AdoptionRecord(writer, instancekey.SourcePlatform, instancekey.Digest("k1")), true},
- {"a stale record from session 5", instancekey.AdoptionRecord(instancekey.Binding{Host: host, Incarnation: "inc-1", Session: 5, Generation: "g5"}, instancekey.SourcePlatform, instancekey.Digest("k1")), false},
- {"a malformed record", []byte(`{"v":1}`), false},
- {"no record", nil, false},
+ {"the releasing writer's record", instancekey.AdoptionRecord(writer, instancekey.SourcePlatform, instancekey.Digest("k1")), true, false},
+ {"a stale record from session 5", instancekey.AdoptionRecord(instancekey.Binding{Host: host, Incarnation: "inc-1", Session: 5, Generation: "g5"}, instancekey.SourcePlatform, instancekey.Digest("k1")), false, false},
+ {"a malformed record", []byte(`{"v":1}`), false, true},
+ {"no record", nil, false, false},
} {
t.Run(tc.name, func(t *testing.T) {
data := release.NewDirStore(t.TempDir())
@@ -52,6 +56,15 @@ func TestReleaseFlushPublishesOnlyTheReleasingWritersAdoption(t *testing.T) {
f := ReleaseFlushFiles{Data: data, Registry: reg, FilesDir: func(string) string { return dir }, EnvPath: func(h string) string { return filepath.Join(envDir, h+".env") }}
obs := hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Host: host, Holder: &hostclaim.Holder{Location: "instances/" + host + "/inc", Incarnation: "inc-1", Session: 6, Generation: "g6"}}}
seq, err := f.Flush(ctx, obs)
+ if tc.refuses {
+ if !errors.Is(err, ErrClaimStopUnavailable) {
+ t.Fatalf("a malformed record was flushed: %v", err)
+ }
+ if _, ok, err := hostfiles.Load(ctx, data, "instances/"+host+"/inc/files/g6/"); ok || err != nil {
+ t.Fatalf("a refused flush committed a manifest: %v %v", ok, err)
+ }
+ return
+ }
if err != nil {
t.Fatal(err)
}
@@ -70,3 +83,29 @@ func TestReleaseFlushPublishesOnlyTheReleasingWritersAdoption(t *testing.T) {
})
}
}
+
+// Fix round 1, M6: a host with no paths (no route, or a route that could
+// not be read) is refused before anything is staged, never flushed with an
+// empty env path read as "no env file".
+func TestReleaseFlushRefusesAHostWithNoPaths(t *testing.T) {
+ ctx := context.Background()
+ const host = "shop.example"
+ data := release.NewDirStore(t.TempDir())
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ paths := relocationHostPaths(t.TempDir(), reg)
+ if p := paths(host); p.FilesDir != "" || p.EnvPath != "" {
+ t.Fatalf("paths without a route: %+v", p)
+ }
+ f := ReleaseFlushFiles{Data: data, Registry: reg, FilesDir: func(h string) string { return paths(h).FilesDir }, EnvPath: func(h string) string { return paths(h).EnvPath }}
+ obs := hostclaim.Observed{Host: host, Claim: &hostclaim.Claim{Host: host, Holder: &hostclaim.Holder{Location: "instances/" + host + "/inc", Incarnation: "inc-1", Session: 6, Generation: "g6"}}}
+ if _, err := f.Flush(ctx, obs); !errors.Is(err, ErrClaimStopUnavailable) {
+ t.Fatalf("flushed with no paths: %v", err)
+ }
+ if _, ok, err := hostfiles.Load(ctx, data, "instances/"+host+"/inc/files/g6/"); ok || err != nil {
+ t.Fatalf("a refused flush committed a manifest: %v %v", ok, err)
+ }
+}
diff --git a/internal/host/relocationsweep.go b/internal/host/relocationsweep.go
index ae550ac6a..fdbd84ea8 100644
--- a/internal/host/relocationsweep.go
+++ b/internal/host/relocationsweep.go
@@ -4,23 +4,53 @@ import (
"context"
"encoding/json"
"errors"
+ "fmt"
+ "log"
"os"
"path/filepath"
"strings"
+ "sync"
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/release"
)
-// sweepRelocationJournals is the lifecycle half of N9 (ruling 61): every
-// region HA journal under <DataDir>/relocate names its host inside and its
-// op in its path, and is removed once the host's claim no longer runs that
-// op (completed and marked discharged, replaced by a successor, or the
-// claim gone). Positions,
-// attempts and install journals are needed only while their op runs;
-// the build scratch only while its op is installing. A journal that
-// cannot be read is left in place and reported: it is census-opaque, so
-// it blocks nothing, and absence is never evidence.
+// relocationOpJournals are the files under relocate/ops/<op>/ that name
+// their host inside: the install and attempt journals and the probe stop's
+// fence owner.
+var relocationOpJournals = []string{"install.json", "attempts.json", "probe-fence.json"}
+
+// relocationStraySaid throttles the sweep's line about an entry under
+// relocate/ that is no relocation journal to once per path per process.
+var relocationStraySaid sync.Map
+
+func sayRelocationStray(path string) {
+ if _, said := relocationStraySaid.LoadOrStore(path, true); !said {
+ log.Printf("relocation sweep: %s is not a relocation journal; left in place for an operator", path)
+ }
+}
+
+// sweepRelocationJournals is the lifecycle half of N9 (ruling 61). Every
+// region HA journal under <DataDir>/relocate names its op in its path and
+// its host inside (the build scratch names its host in its file name): the
+// install, attempt and probe-fence journals under ops/<op>/, the release
+// positions at release/<op>.json and the build's scratch under build/<op>/.
+// Each is removed once the host's claim no longer runs that op: replaced
+// by a successor, the claim gone, or completed and marked discharged (a
+// completed op's journals stay until that mark, ruling 76). While the op
+// runs, every one of them stays.
+//
+// A journal that cannot be read is left in place and reported, and so is
+// a claim whose completed relocation cannot be read: both are
+// census-opaque, so they block nothing, and absence is never evidence.
+// What no committed journal can be is removed: an ops directory holding
+// no host-bearing journal, an atomic write's leftover temporary under
+// release/, a build directory left empty. An entry the sweep does not know
+// is left in place and logged once.
+//
+// The caller runs it once per pass, between the pass's own steps, never
+// beside a relocation journal write: an ops directory and a temporary are
+// only ever a write's leftovers then.
func sweepRelocationJournals(ctx context.Context, dataDir string, claims release.Store) error {
root := filepath.Join(dataDir, "relocate")
var errs []error
@@ -35,7 +65,13 @@ func sweepRelocationJournals(ctx context.Context, dataDir string, claims release
// Ruling 76: a completed relocation's journals, the probe stop's
// fence owner among them, stay until the box it completed on has
// discharged, released its hold, lifted its fence and marked it.
- if done, ok := hostclaim.ReadCompletedRelocation(obs.Claim); ok && done.Op == op && !done.Discharged {
+ // A record that cannot be read keeps every journal: it may be this
+ // op's, and absence is never evidence.
+ done, ok, err := hostclaim.ReadCompletedRelocation(obs.Claim)
+ if err != nil {
+ return false, fmt.Errorf("%s: %w", host, err)
+ }
+ if ok && done.Op == op && !done.Discharged {
return false, nil
}
return obs.Claim.Op.ID != op, nil
@@ -50,87 +86,139 @@ func sweepRelocationJournals(ctx context.Context, dataDir string, claims release
Request struct{ Host string }
}
if err := json.Unmarshal(raw, &probe); err != nil {
- return "", err
+ return "", fmt.Errorf("%s: %w", path, err)
}
if probe.Host != "" {
return probe.Host, nil
}
- return probe.Request.Host, nil
+ if probe.Request.Host != "" {
+ return probe.Request.Host, nil
+ }
+ return "", fmt.Errorf("%s names no host", path)
}
- // ops/<op>/{install,attempts}.json
+ // ops/<op>/{install,attempts,probe-fence}.json
ops, err := os.ReadDir(filepath.Join(root, "ops"))
if err != nil && !errors.Is(err, os.ErrNotExist) {
return err
}
for _, e := range ops {
dir := filepath.Join(root, "ops", e.Name())
- host, readErr := "", error(nil)
- for _, name := range []string{"install.json", "attempts.json", "probe-fence.json"} {
+ if !e.IsDir() {
+ sayRelocationStray(dir)
+ continue
+ }
+ host, found, readErr := "", false, error(nil)
+ for _, name := range relocationOpJournals {
h, err := hostOf(filepath.Join(dir, name))
- if err == nil && h != "" {
+ if errors.Is(err, os.ErrNotExist) {
+ continue
+ }
+ found = true
+ if err == nil {
host = h
break
}
- if err != nil && !errors.Is(err, os.ErrNotExist) {
- readErr = errors.Join(readErr, err)
- }
- }
- if host == "" {
- errs = append(errs, readErr)
- continue
+ readErr = errors.Join(readErr, err)
}
- if done, err := gone(host, e.Name()); err != nil {
- errs = append(errs, err)
- } else if done {
+ switch {
+ case host != "":
+ if done, err := gone(host, e.Name()); err != nil {
+ errs = append(errs, err)
+ } else if done {
+ errs = append(errs, os.RemoveAll(dir))
+ }
+ case found:
+ errs = append(errs, readErr) // unreadable: kept and reported
+ default:
+ // No host-bearing journal was ever committed here: what is
+ // left is a write's temporary, which no op can need.
errs = append(errs, os.RemoveAll(dir))
}
}
- // release/<op>.json
+ // release/<op>.json, and an atomic write's leftover temporaries.
positions, err := os.ReadDir(filepath.Join(root, "release"))
if err != nil && !errors.Is(err, os.ErrNotExist) {
- return err
+ return errors.Join(append(errs, err)...)
}
for _, e := range positions {
- op, ok := strings.CutSuffix(e.Name(), ".json")
- if !ok {
- continue
- }
path := filepath.Join(root, "release", e.Name())
- host, err := hostOf(path)
- if err != nil || host == "" {
- errs = append(errs, err)
- continue
- }
- if done, err := gone(host, op); err != nil {
- errs = append(errs, err)
- } else if done {
+ op, isJSON := strings.CutSuffix(e.Name(), ".json")
+ switch {
+ case e.IsDir():
+ sayRelocationStray(path)
+ case isJSON:
+ host, err := hostOf(path)
+ if err != nil {
+ errs = append(errs, err)
+ continue
+ }
+ if done, err := gone(host, op); err != nil {
+ errs = append(errs, err)
+ } else if done {
+ errs = append(errs, os.Remove(path))
+ }
+ case relocationWriteLeftover(e.Name()):
errs = append(errs, os.Remove(path))
+ default:
+ sayRelocationStray(path)
}
}
- // build/<op>/<host>.db: the scratch of a build whose op is over.
+ // build/<op>/<host>.db and its SQLite sidecars: the scratch of a build
+ // whose op is over.
builds, err := os.ReadDir(filepath.Join(root, "build"))
if err != nil && !errors.Is(err, os.ErrNotExist) {
return errors.Join(append(errs, err)...)
}
for _, e := range builds {
dir := filepath.Join(root, "build", e.Name())
+ if !e.IsDir() {
+ sayRelocationStray(dir)
+ continue
+ }
files, err := os.ReadDir(dir)
if err != nil {
errs = append(errs, err)
continue
}
+ host := ""
for _, f := range files {
- host, ok := strings.CutSuffix(f.Name(), ".db")
- if !ok {
- continue
+ if h, ok := buildScratchHost(f.Name()); ok {
+ host = h
+ break
}
+ }
+ switch {
+ case host != "":
if done, err := gone(host, e.Name()); err != nil {
errs = append(errs, err)
} else if done {
errs = append(errs, os.RemoveAll(dir))
}
- break
+ case len(files) == 0:
+ errs = append(errs, os.Remove(dir))
+ default:
+ sayRelocationStray(dir)
}
}
return errors.Join(errs...)
}
+
+// relocationWriteLeftover is a temporary an interrupted atomic write left
+// under release/: writeFileAtomic's ".<op>.json.<random>", or a fixed
+// "<op>.json.tmp".
+func relocationWriteLeftover(name string) bool {
+ return strings.HasSuffix(name, ".json.tmp") || (strings.HasPrefix(name, ".") && strings.Contains(name, ".json."))
+}
+
+// buildScratchHost is the host a build scratch file names: <host>.db, or
+// one of SQLite's sidecars beside it (-wal, -shm, -journal).
+func buildScratchHost(name string) (string, bool) {
+ for _, sidecar := range []string{"-wal", "-shm", "-journal"} {
+ if base, ok := strings.CutSuffix(name, sidecar); ok {
+ name = base
+ break
+ }
+ }
+ host, ok := strings.CutSuffix(name, ".db")
+ return host, ok && host != ""
+}
diff --git a/internal/host/relocationsweep_test.go b/internal/host/relocationsweep_test.go
index 4f36ce4cd..83e4f14e4 100644
--- a/internal/host/relocationsweep_test.go
+++ b/internal/host/relocationsweep_test.go
@@ -3,6 +3,7 @@ package host
import (
"encoding/json"
"errors"
+ "log"
"os"
"path/filepath"
"strings"
@@ -153,3 +154,121 @@ func TestRelocationJournalsOfACompletedOpStayUntilItsDischarge(t *testing.T) {
t.Fatalf("the discharged op's journals survived: %v", err)
}
}
+
+// sweepClaim is shop.example's claim, adopted on "dst", with its op
+// replaced by op.
+func sweepClaim(t *testing.T, op hostclaim.Op) *release.DirStore {
+ t.Helper()
+ ctx := t.Context()
+ store := release.NewDirStore(t.TempDir())
+ created := time.Date(2026, 10, 7, 0, 0, 0, 0, time.UTC)
+ record := provision.Record{Schema: 1, Host: "shop.example", Slug: "shop", App: "app", Account: "acct", Region: "ie", Channel: "stable", CreatedAt: created}
+ if err := provision.CreateRecord(ctx, store, record); err != nil {
+ t.Fatal(err)
+ }
+ obs, err := hostclaim.Read(ctx, store, record.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ obs, err = hostclaim.Update(ctx, store, obs, hostclaim.Adopt(hostclaim.NewHolder{Family: "instance", Key: provision.RecordKey(record.Account, record.App, record.Host),
+ Account: record.Account, App: record.App, Host: record.Host, Region: record.Region, CreatedAt: created}, "dst", nil, nil, "adopt", func() string { return "id" }))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := hostclaim.Update(ctx, store, obs, func(c *hostclaim.Claim) error { c.Op = op; return nil }); err != nil {
+ t.Fatal(err)
+ }
+ return store
+}
+
+func writeSweepFiles(t *testing.T, data string, files map[string]string) {
+ t.Helper()
+ for path, body := range files {
+ full := filepath.Join(data, filepath.FromSlash(path))
+ if err := os.MkdirAll(filepath.Dir(full), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(full, []byte(body), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+// Fix round 1, I1: a serving claim whose completed relocation cannot be
+// read keeps every journal of its host, the probe stop's fence owner
+// included, and the sweep reports it: it may be that op's, and absence is
+// never evidence.
+func TestRelocationJournalsStayWhileTheCompletionCannotBeRead(t *testing.T) {
+ for name, raw := range map[string]string{"unparseable": `{"op":"rel-done",`, "no box": `{"op":"rel-done","stop":{"session":12}}`} {
+ t.Run(name, func(t *testing.T) {
+ store := sweepClaim(t, hostclaim.Op{Kind: hostclaim.KindServing, Exec: "dst", Participants: []string{"dst"}, Journal: map[string]string{hostclaim.JournalCompleted: raw}})
+ data := t.TempDir()
+ files := map[string]string{
+ "relocate/ops/rel-done/install.json": `{"v":1,"host":"shop.example","op":"rel-done"}`,
+ "relocate/ops/rel-done/probe-fence.json": `{"host":"shop.example","session":12,"owner":"0123"}`,
+ "relocate/release/rel-done.json": `{"request":{"Host":"shop.example"}}`,
+ "relocate/build/rel-done/shop.example.db": "scratch",
+ }
+ writeSweepFiles(t, data, files)
+ if err := sweepRelocationJournals(t.Context(), data, store); !errors.Is(err, hostclaim.ErrRelocation) {
+ t.Fatalf("an unreadable completion was not reported: %v", err)
+ }
+ for path := range files {
+ if _, err := os.Stat(filepath.Join(data, filepath.FromSlash(path))); err != nil {
+ t.Errorf("%s was swept over an unreadable completion: %v", path, err)
+ }
+ }
+ })
+ }
+}
+
+// Fix round 1, M5: what no committed journal can be is removed (an ops
+// directory with no host-bearing journal, an atomic write's leftover
+// temporaries under release/, a build directory holding only SQLite
+// sidecars of an ended op, or nothing); the live op's sidecar stays; an
+// entry the sweep does not know stays and is logged once, not every pass,
+// and is no error.
+func TestRelocationSweepRemovesLeftoversAndLogsAStrayOnce(t *testing.T) {
+ store := sweepClaim(t, hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "rel-live", Phase: hostclaim.PhaseInstalling, Exec: "dst", Participants: []string{"src", "dst"},
+ Intent: map[string]string{"region": "ie", "source_box": "src", "target_box": "dst"}})
+ data := t.TempDir()
+ writeSweepFiles(t, data, map[string]string{
+ "relocate/ops/rel-tmp/.install.json.123456": `{"v":1,"host":`,
+ "relocate/release/rel-x.json.tmp": `{"request":`,
+ "relocate/release/.rel-y.json.987654": `{"req`,
+ "relocate/build/rel-wal/shop.example.db-wal": "wal",
+ "relocate/build/rel-jrn/shop.example.db-journal": "journal",
+ "relocate/build/rel-live/shop.example.db-wal": "wal",
+ "relocate/build/stray.txt": "?",
+ "relocate/release/notes.txt": "?",
+ })
+ for _, dir := range []string{"relocate/ops/rel-empty", "relocate/build/rel-empty"} {
+ if err := os.MkdirAll(filepath.Join(data, filepath.FromSlash(dir)), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ }
+ var logged strings.Builder
+ log.SetOutput(&logged)
+ t.Cleanup(func() { log.SetOutput(os.Stderr) })
+ for pass := 0; pass < 2; pass++ {
+ if err := sweepRelocationJournals(t.Context(), data, store); err != nil {
+ t.Fatalf("pass %d: %v", pass, err)
+ }
+ }
+ for _, gone := range []string{"relocate/ops/rel-tmp", "relocate/ops/rel-empty", "relocate/release/rel-x.json.tmp", "relocate/release/.rel-y.json.987654",
+ "relocate/build/rel-wal", "relocate/build/rel-jrn", "relocate/build/rel-empty"} {
+ if _, err := os.Stat(filepath.Join(data, filepath.FromSlash(gone))); !errors.Is(err, os.ErrNotExist) {
+ t.Errorf("%s survived the sweep: %v", gone, err)
+ }
+ }
+ for _, kept := range []string{"relocate/build/rel-live/shop.example.db-wal", "relocate/build/stray.txt", "relocate/release/notes.txt"} {
+ if _, err := os.Stat(filepath.Join(data, filepath.FromSlash(kept))); err != nil {
+ t.Errorf("%s was removed: %v", kept, err)
+ }
+ }
+ for _, stray := range []string{"stray.txt", "notes.txt"} {
+ if n := strings.Count(logged.String(), stray); n != 1 {
+ t.Errorf("%s logged %d times over two passes, want once: %q", stray, n, logged.String())
+ }
+ }
+}
diff --git a/internal/hostclaim/relocation.go b/internal/hostclaim/relocation.go
index b6b9d3e03..9d9d71348 100644
--- a/internal/hostclaim/relocation.go
+++ b/internal/hostclaim/relocation.go
@@ -165,9 +165,11 @@ type LocalInstall struct {
type Probe struct {
Box string `json:"box"`
Session uint64 `json:"session"`
- // At is when the probe passed (rev 4, ruling 67): the start of the
- // bound on obtaining its stop receipt. Zero in a probe journaled before
- // rev 4; the worker then counts from probing_since.
+ // At is when the probe passed (rev 4, ruling 67): the worker marks the
+ // op stuck ReleaseDeadline after At if no stop receipt is journaled.
+ // Zero in a probe journaled before rev 4: the worker then takes the
+ // probe as passed at probing_since plus ProbeDeadline, the latest it
+ // can have passed, and counts ReleaseDeadline from there.
At time.Time `json:"at,omitzero"`
}
@@ -1080,7 +1082,7 @@ func CompleteRelocation(opID string, newID func() string) func(*Claim) error {
if c.Op.Journal["conversion"] == "committed-landing" {
// M1 on amadan #236: the move this relocation finished is named
// on the serving op, so moves' close can find it.
- next.Journal["released_move"] = opID
+ next.Journal[journalReleasedMove] = opID
}
c.Op = next
return nil
diff --git a/internal/hostclaim/relocation_probestop.go b/internal/hostclaim/relocation_probestop.go
index 31ebf7051..a2fa9c528 100644
--- a/internal/hostclaim/relocation_probestop.go
+++ b/internal/hostclaim/relocation_probestop.go
@@ -1,6 +1,7 @@
package hostclaim
import (
+ "encoding/json"
"fmt"
"slices"
"time"
@@ -74,13 +75,28 @@ type CompletedRelocation struct {
// JournalCompleted is the serving op's journal key for CompletedRelocation.
const JournalCompleted = "completed_relocation"
-// ReadCompletedRelocation reads the serving op's record; false when none.
-func ReadCompletedRelocation(c *Claim) (CompletedRelocation, bool) {
+// ReadCompletedRelocation reads the serving op's record. ok is false, with
+// no error, only when there is none: the op is not serving, or its journal
+// has no JournalCompleted entry. An entry that is present but does not
+// decode, or names no op, box or session, is an error, never "none"
+// (absence is never evidence): the barrier refuses over it and the sweep
+// keeps the op's journals.
+func ReadCompletedRelocation(c *Claim) (CompletedRelocation, bool, error) {
+ if c == nil || c.Op.Kind != KindServing {
+ return CompletedRelocation{}, false, nil
+ }
+ raw, present := c.Op.Journal[JournalCompleted]
+ if !present {
+ return CompletedRelocation{}, false, nil
+ }
var r CompletedRelocation
- if c == nil || c.Op.Kind != KindServing || !readJournal(c.Op, JournalCompleted, &r) || r.Op == "" || r.Stop.Box == "" || r.Stop.Session == 0 {
- return CompletedRelocation{}, false
+ if err := json.Unmarshal([]byte(raw), &r); err != nil {
+ return CompletedRelocation{}, false, fmt.Errorf("%w: the completed relocation cannot be read: %v", ErrRelocation, err)
+ }
+ if r.Op == "" || r.Stop.Box == "" || r.Stop.Session == 0 {
+ return CompletedRelocation{}, false, fmt.Errorf("%w: the completed relocation names no op, box or session", ErrRelocation)
}
- return r, true
+ return r, true, nil
}
// completionBarrier is region HA ruling 77, the one barrier every start
@@ -91,9 +107,14 @@ func ReadCompletedRelocation(c *Claim) (CompletedRelocation, bool) {
// successor operation would answer ErrLost, so nothing starts. A
// completion box on lost, or the one the start itself cites as fenced
// (fencedSource, a failover's source), holds nothing a successor needs: a
-// dead box's slot binds nothing on another box.
+// dead box's slot binds nothing on another box. A record that cannot be
+// read refuses every start, a failover's included: which box it binds is
+// unknown.
func completionBarrier(c *Claim, fencedSource string) error {
- done, ok := ReadCompletedRelocation(c)
+ done, ok, err := ReadCompletedRelocation(c)
+ if err != nil {
+ return err
+ }
if !ok || done.Discharged || slices.Contains(c.Lost, done.Stop.Box) || (fencedSource != "" && fencedSource == done.Stop.Box) {
return nil
}
@@ -104,10 +125,13 @@ func completionBarrier(c *Claim, fencedSource string) error {
// relocation opID completed on has discharged its participation. It
// changes nothing else; a repeat is ErrAlreadyRecorded. A serving op with
// no completed relocation is ErrWrongState, and one that records another
-// op is ErrRelocation.
+// op, or a record that cannot be read, is ErrRelocation.
func MarkCompletedDischarged(opID string) func(*Claim) error {
return func(c *Claim) error {
- r, ok := ReadCompletedRelocation(c)
+ r, ok, err := ReadCompletedRelocation(c)
+ if err != nil {
+ return err
+ }
if !ok {
return fmt.Errorf("%w: no completed relocation on the serving op", ErrWrongState)
}
diff --git a/internal/hostclaim/relocation_probestop_test.go b/internal/hostclaim/relocation_probestop_test.go
index 9ce573b59..914e4f792 100644
--- a/internal/hostclaim/relocation_probestop_test.go
+++ b/internal/hostclaim/relocation_probestop_test.go
@@ -108,7 +108,7 @@ func TestCompleteRelocationRequiresTheProbeStop(t *testing.T) {
t.Fatalf("a second, different stop: %v", err)
}
o = apply(CompleteRelocation("rel-1", counter()))
- done, ok := ReadCompletedRelocation(o.Claim)
+ done, ok, _ := ReadCompletedRelocation(o.Claim)
if !ok || done.Op != "rel-1" || done.Stop != stop || done.Stamp != o.Claim.Holder.Stamp || done.Phase != PhaseRelocatingProbing || done.Generation != "P1" {
t.Fatalf("completed relocation: %+v %v", done, ok)
}
@@ -127,7 +127,7 @@ func TestCompleteRelocationJournalsReleasedMoveForAConvertedLanding(t *testing.T
if o.Claim.Op.Journal["released_move"] != "rel-1" {
t.Fatalf("serving op journal: %v", o.Claim.Op.Journal)
}
- if _, ok := ReadCompletedRelocation(o.Claim); !ok {
+ if _, ok, err := ReadCompletedRelocation(o.Claim); !ok || err != nil {
t.Fatal("no completed relocation beside the released move")
}
}
@@ -190,7 +190,7 @@ func TestRelocationSuccessorWaitsForTheCompletedDischarge(t *testing.T) {
if _, err := Update(ctx, s, o, MarkCompletedDischarged("rel-1")); !errors.Is(err, ErrAlreadyRecorded) {
t.Fatalf("a repeat: %v", err)
}
- if done, ok := ReadCompletedRelocation(o.Claim); !ok || !done.Discharged || done.Stop != stopOf(o) {
+ if done, ok, err := ReadCompletedRelocation(o.Claim); err != nil || !ok || !done.Discharged || done.Stop != stopOf(o) {
t.Fatalf("the mark changed the record: %+v", done)
}
next, err := StartRelocation(ctx, s, o, successor(ReasonDrain, nil), relAuthority{})
@@ -227,7 +227,7 @@ func completedDrain(t *testing.T) (release.Store, Observed) {
s, o := probedDrainIn(t)
o = step(t, s, o, RecordProbeStop("rel-1", stopOf(o)))
o = step(t, s, o, CompleteRelocation("rel-1", counter()))
- if done, ok := ReadCompletedRelocation(o.Claim); !ok || done.Discharged || o.Claim.Op.Exec != relTarget {
+ if done, ok, err := ReadCompletedRelocation(o.Claim); err != nil || !ok || done.Discharged || o.Claim.Op.Exec != relTarget {
t.Fatalf("A to B did not complete undischarged: %+v %v", done, ok)
}
return s, o
@@ -265,3 +265,48 @@ func TestMoveSuccessorWaitsForTheCompletedDischarge(t *testing.T) {
t.Fatalf("move op %+v", next.Claim.Op)
}
}
+
+// Fix round 1, I1: a completed_relocation entry that is present but cannot
+// be read is an error, never "none". Every start from serving refuses over
+// it (a failover citing a fence included, since which box the record binds
+// is unknown), the discharge mark refuses, and the claim is left as it is.
+func TestAnUnreadableCompletedRelocationRefusesEveryStart(t *testing.T) {
+ ctx := context.Background()
+ for name, raw := range map[string]string{
+ "unparseable": `{"op":"rel-1","stop":`,
+ "no box": `{"op":"rel-1","stop":{"session":4,"receipt":"fence-1"}}`,
+ "no session": `{"op":"rel-1","stop":{"box":"` + relTarget + `","receipt":"fence-1"}}`,
+ "no op": `{"stop":{"box":"` + relTarget + `","session":4,"receipt":"fence-1"}}`,
+ "empty": ``,
+ "another shape": `["rel-1"]`,
+ } {
+ t.Run(name, func(t *testing.T) {
+ s, o := completedDrain(t)
+ o = step(t, s, o, func(c *Claim) error { c.Op.Journal[JournalCompleted] = raw; return nil })
+ if _, ok, err := ReadCompletedRelocation(o.Claim); ok || !errors.Is(err, ErrRelocation) {
+ t.Fatalf("read as %v, %v", ok, err)
+ }
+ successor := RelocationStart{OpID: "rel-2", By: "worker", Now: relObserved.Add(time.Hour), Region: "ie", Reason: ReasonDrain,
+ SourceBox: relTarget, SourceEnrollment: relEnrollment(relTarget), TargetBox: relSpare, TargetEnrollment: relEnrollment(relSpare),
+ DirectoryRevision: 9, AdmissionID: "adm-2"}
+ if _, err := StartRelocation(ctx, s, o, successor, relAuthority{}); !errors.Is(err, ErrRelocation) {
+ t.Fatalf("a drain started over an unreadable completion: %v", err)
+ }
+ ref := putFence(t, s, relTarget)
+ failover := successor
+ failover.Reason, failover.Fence = ReasonFailover, &ref
+ if _, err := StartRelocation(ctx, s, o, failover, relAuthority{}); !errors.Is(err, ErrRelocation) {
+ t.Fatalf("a failover started over an unreadable completion: %v", err)
+ }
+ if err := StartGuard(copyOf(t, o.Claim), []string{"uk"}, []string{relSpare}); !errors.Is(err, ErrRelocation) {
+ t.Fatalf("StartGuard: %v", err)
+ }
+ if err := MarkCompletedDischarged("rel-1")(copyOf(t, o.Claim)); !errors.Is(err, ErrRelocation) {
+ t.Fatalf("the discharge mark over an unreadable completion: %v", err)
+ }
+ if now, err := Read(ctx, s, o.Claim.Host); err != nil || now.ETag != o.ETag {
+ t.Fatalf("a refused start wrote the claim: %v", err)
+ }
+ })
+ }
+}
diff --git a/internal/hostclaim/relocation_test.go b/internal/hostclaim/relocation_test.go
index 783ce9d3e..2359995d4 100644
--- a/internal/hostclaim/relocation_test.go
+++ b/internal/hostclaim/relocation_test.go
@@ -279,7 +279,7 @@ func TestRelocationFailoverCitesTheFenceAndDeclaresTheLoss(t *testing.T) {
l.Session != sourceSession || l.Generation != "g1" || l.Discharge.Qualified != qualified || l.Timers != TimersUnknown || l.Reported {
t.Fatalf("retained loss: %+v", l)
}
- done, ok := ReadCompletedRelocation(c)
+ done, ok, _ := ReadCompletedRelocation(c)
if len(c.Op.Journal) != 1 || !ok || done.Op != "rel-1" || done.Stop.Session != c.Holder.Session || done.Phase != PhaseRelocatingProbing || !slices.Contains(c.Lost, relSource) {
t.Fatalf("serving kept the journal, or lost the box: %+v %v", c.Op, c.Lost)
}
diff --git a/internal/hostrelocate/participant_gate_test.go b/internal/hostrelocate/participant_gate_test.go
index 273a9b467..ddec620a5 100644
--- a/internal/hostrelocate/participant_gate_test.go
+++ b/internal/hostrelocate/participant_gate_test.go
@@ -161,3 +161,52 @@ func TestParticipantRecordsNoProbeStopWhileRootRefuses(t *testing.T) {
t.Fatalf("the stop once root answered: %v", err)
}
}
+
+// Fix round 1, M4 (ruling 62): sameInstallation accepts a fresh read that
+// differs only in an effect's own bookkeeping, and refuses one whose
+// installation moved: another stamp, install stamp, prepared P, session
+// or phase, or a stuck mark. (The claim store itself refuses a changed
+// incarnation, or an installed op's executor moved off its target.)
+func TestSameInstallationRefusesAMovedInstallation(t *testing.T) {
+ ctx := context.Background()
+ installed := func(t *testing.T) (*fixture, hostclaim.Observed, Participant) {
+ f := newFixture(t)
+ releasedAndInstalling(t, f)
+ f.worker(ErrNotYet) // P is not built yet
+ p := f.claim().Claim.Op.Journal["prepared"]
+ f.participant(func(c *hostclaim.Claim) error { c.Op.Journal["test.built"] = p; return nil })
+ if o := f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseInstalled {
+ t.Fatalf("not installed: %s", o.Claim.Op.Phase)
+ }
+ return f, f.claim(), Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: &fakeBox{f: f, box: "dst"}, Now: f.clock}
+ }
+ t.Run("the effect's own bookkeeping", func(t *testing.T) {
+ f, obs, p := installed(t)
+ f.participant(func(c *hostclaim.Claim) error { c.Op.Journal["test.inventory"] = "dst"; return nil })
+ fresh, err := p.sameInstallation(ctx, obs, hostclaim.PhaseInstalled)
+ if err != nil || fresh.ETag == obs.ETag || fresh.Claim.Op.Journal["test.inventory"] != "dst" {
+ t.Fatalf("a fresh read of the same installation: %v", err)
+ }
+ })
+ for name, change := range map[string]func(*hostclaim.Claim){
+ "another stamp": func(c *hostclaim.Claim) { c.Holder.Stamp++ },
+ "another install stamp": func(c *hostclaim.Claim) { c.Op.Journal["install_stamp"] += "0" },
+ "another prepared P": func(c *hostclaim.Claim) { c.Op.Journal["prepared"] = "P9" },
+ "another session": func(c *hostclaim.Claim) { c.Holder.Session++ },
+ "stuck": func(c *hostclaim.Claim) { c.Op.Journal["stuck"] = "operator" },
+ } {
+ t.Run(name, func(t *testing.T) {
+ f, obs, p := installed(t)
+ f.participant(func(c *hostclaim.Claim) error { change(c); return nil })
+ if _, err := p.sameInstallation(ctx, obs, hostclaim.PhaseInstalled); !errors.Is(err, ErrRefused) || !errors.Is(err, ErrNotYet) {
+ t.Fatalf("a moved installation was accepted: %v", err)
+ }
+ })
+ }
+ t.Run("another phase", func(t *testing.T) {
+ _, obs, p := installed(t)
+ if _, err := p.sameInstallation(ctx, obs, hostclaim.PhaseRelocatingProbing); !errors.Is(err, ErrRefused) {
+ t.Fatalf("another phase was accepted: %v", err)
+ }
+ })
+}
diff --git a/internal/hostrelocate/participant_test.go b/internal/hostrelocate/participant_test.go
index 06d151f40..7ccba9449 100644
--- a/internal/hostrelocate/participant_test.go
+++ b/internal/hostrelocate/participant_test.go
@@ -65,7 +65,8 @@ func (b *fakeBox) Finish(ctx context.Context, h, op string) error {
if err != nil {
return err
}
- if done, ok := hostclaim.ReadCompletedRelocation(obs.Claim); ok && done.Op == op && done.Stop.Box == b.box && !done.Discharged {
+ done, ok, err := hostclaim.ReadCompletedRelocation(obs.Claim)
+ if err == nil && ok && done.Op == op && done.Stop.Box == b.box && !done.Discharged {
_, err = hostclaim.Update(ctx, b.f.s, obs, hostclaim.MarkCompletedDischarged(op))
}
return err
diff --git a/internal/hostrelocate/worker.go b/internal/hostrelocate/worker.go
index cd447030e..c49786246 100644
--- a/internal/hostrelocate/worker.go
+++ b/internal/hostrelocate/worker.go
@@ -406,9 +406,11 @@ func (w Worker) Step(ctx context.Context, host string) (hostclaim.Observed, erro
// 67: bounded. If root's stop receipt cannot be had (the
// execution lease lost after RecordProbe, so the participant
// never passes Ready; a stop root keeps refusing), the op is
- // stuck ReleaseDeadline after the probe passed, for an operator.
- // Nothing is released: the participation's denial, the row's
- // hold and the session stay as they are.
+ // stuck ReleaseDeadline after the probe passed (Probe.At; for an
+ // untimed probe journaled before rev 4, probing_since plus
+ // ProbeDeadline, the latest it can have passed), for an
+ // operator. Nothing is released: the participation's denial,
+ // the row's hold and the session stay as they are.
var p hostclaim.Probe
passed := since.Add(ProbeDeadline)
if json.Unmarshal([]byte(c.Op.Journal["probe"]), &p) == nil && !p.At.IsZero() {
@@ -435,8 +437,13 @@ func (w Worker) Step(ctx context.Context, host string) (hostclaim.Observed, erro
// ProbeDeadline (a retired stamp, a participation main denies) is
// stuck for an operator: no automatic replacement of a live target
// is made here.
+ // A missing or unreadable installed_at is an error, never "no
+ // bound" (absence is never evidence), as probing_since is above.
at, err := time.Parse(time.RFC3339Nano, c.Op.Journal["installed_at"])
- if err == nil && w.Now().Sub(at) > ProbeDeadline {
+ if err != nil {
+ return obs, fmt.Errorf("installed_at of %s: %w", id, err)
+ }
+ if w.Now().Sub(at) > ProbeDeadline {
return w.update(ctx, obs, hostclaim.MarkStuck(id, fmt.Sprintf("%s did not enter probing within %s of installed", c.Op.Exec, ProbeDeadline)))
}
return obs, ErrNotYet
diff --git a/internal/hostrelocate/worker_bounds_test.go b/internal/hostrelocate/worker_bounds_test.go
index 71dd1034f..0b4c4bd28 100644
--- a/internal/hostrelocate/worker_bounds_test.go
+++ b/internal/hostrelocate/worker_bounds_test.go
@@ -3,6 +3,7 @@ package hostrelocate
import (
"context"
"encoding/json"
+ "errors"
"strings"
"testing"
"time"
@@ -127,3 +128,56 @@ func TestWorkerBoundsARefusedProbe(t *testing.T) {
t.Fatalf("not stuck on dst for an operator: %+v", o.Claim.Op)
}
}
+
+// Fix round 1, M3: a timed probe's bound counts from Probe.At, not from
+// probing_since and not from probing_since plus ProbeDeadline. The probe
+// passes 7 minutes into probing: ReleaseDeadline after At the op waits,
+// a second later it is stuck.
+func TestWorkerBoundsATimedProbeFromItsPassNotProbingSince(t *testing.T) {
+ f := newFixture(t)
+ id := releasedAndInstalling(t, f)
+ f.installAndProbe("dst") // probing_since is f.now
+ passed := f.now.Add(7 * time.Minute)
+ o := f.claim()
+ f.participant(hostclaim.RecordProbe(id, hostclaim.Probe{Box: "dst", Session: o.Claim.Holder.Session, At: passed}))
+ f.now = passed.Add(ReleaseDeadline)
+ if o := f.worker(ErrNotYet); o.Claim.Op.Journal["stuck"] != "" {
+ t.Fatal("stuck before ReleaseDeadline after the probe passed: the bound counted from probing_since")
+ }
+ f.now = f.now.Add(time.Second)
+ if o := f.worker(nil); o.Claim.Op.Journal["stuck"] == "" {
+ t.Fatalf("not stuck ReleaseDeadline after the probe passed: %+v", o.Claim.Op)
+ }
+}
+
+// Fix round 1, M2: installed's bound needs its start. A missing or
+// unreadable installed_at is an error, never a pass with no bound; the op
+// is neither stuck nor waiting silently.
+func TestWorkerRefusesAnInstalledWithoutItsStart(t *testing.T) {
+ for name, value := range map[string]*string{"missing": nil, "unreadable": new("yesterday")} {
+ t.Run(name, func(t *testing.T) {
+ f := newFixture(t)
+ releasedAndInstalling(t, f)
+ f.worker(ErrNotYet) // P is not built yet
+ p := f.claim().Claim.Op.Journal["prepared"]
+ f.participant(func(c *hostclaim.Claim) error { c.Op.Journal["test.built"] = p; return nil })
+ f.worker(nil) // installed
+ f.participant(func(c *hostclaim.Claim) error {
+ if value == nil {
+ delete(c.Op.Journal, "installed_at")
+ } else {
+ c.Op.Journal["installed_at"] = *value
+ }
+ return nil
+ })
+ f.now = f.now.Add(ProbeDeadline + time.Hour)
+ o, err := f.w.Step(context.Background(), host)
+ if err == nil || errors.Is(err, ErrNotYet) {
+ t.Fatalf("an installed without its start: %v", err)
+ }
+ if o.Claim.Op.Journal["stuck"] != "" || f.claim().Claim.Op.Phase != hostclaim.PhaseInstalled {
+ t.Fatalf("the op moved: %+v", f.claim().Claim.Op)
+ }
+ })
+ }
+}
diff --git a/internal/lease/lease.go b/internal/lease/lease.go
index 61788e219..24c109ac1 100644
--- a/internal/lease/lease.go
+++ b/internal/lease/lease.go
@@ -392,6 +392,21 @@ func (r *Replicator) SyncAndWait(ctx context.Context) error {
return r.db.SyncAndWait(ctx)
}
+// Position is the replicator's own position: the newest TXID litestream
+// has turned into a local level-0 LTX file (litestream's DB.Pos, which
+// keeps the newest L0 file through retention). After a SyncAndWait that
+// returned nil the replica holds everything up to it. Zero, with a nil
+// error, only when litestream holds no LTX file at all: nothing was ever
+// committed through this replicator's database. An unreadable LTX
+// directory or file is an error, never zero.
+func (r *Replicator) Position() (uint64, error) {
+ pos, err := r.db.Pos()
+ if err != nil {
+ return 0, err
+ }
+ return uint64(pos.TXID), nil
+}
+
// Err reports whether lease renewal has failed (leadership lost).
func (r *Replicator) Err() error { return r.rn.Err() }
diff --git a/internal/lease/position_test.go b/internal/lease/position_test.go
new file mode 100644
index 000000000..1c9e34eba
--- /dev/null
+++ b/internal/lease/position_test.go
@@ -0,0 +1,63 @@
+package lease
+
+import (
+ "database/sql"
+ "path/filepath"
+ "testing"
+
+ "github.com/benbjohnson/litestream/file"
+ _ "modernc.org/sqlite"
+)
+
+// Fix round 1 (region HA I2): Position is the replicator's own position.
+// It is zero while litestream holds no LTX file, and after commits and a
+// SyncAndWait it names the newest committed TXID, which the replica holds.
+func TestReplicatorPositionIsItsOwnNewestLTX(t *testing.T) {
+ ctx := t.Context()
+ dir := t.TempDir()
+ path := filepath.Join(dir, "app.db")
+ h := &Held{replicaClient: file.NewReplicaClient(filepath.Join(dir, "replica"))}
+ store, db, err := h.newStore(path, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r := &Replicator{store: store, db: db}
+ if pos, err := r.Position(); err != nil || pos != 0 {
+ t.Fatalf("before any LTX: %d %v", pos, err)
+ }
+ conn, err := sql.Open("sqlite", path+"?_pragma=journal_mode(WAL)")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer conn.Close()
+ if _, err := conn.Exec(`CREATE TABLE t (v INTEGER)`); err != nil {
+ t.Fatal(err)
+ }
+ if err := store.Open(ctx); err != nil {
+ t.Fatal(err)
+ }
+ defer store.Close(ctx)
+ if _, err := conn.Exec(`INSERT INTO t VALUES (1)`); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.SyncAndWait(ctx); err != nil {
+ t.Fatal(err)
+ }
+ first, err := r.Position()
+ if err != nil || first == 0 {
+ t.Fatalf("after a commit: %d %v", first, err)
+ }
+ if _, err := conn.Exec(`INSERT INTO t VALUES (2)`); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.SyncAndWait(ctx); err != nil {
+ t.Fatal(err)
+ }
+ second, err := r.Position()
+ if err != nil || second <= first {
+ t.Fatalf("after a second commit: %d (was %d) %v", second, first, err)
+ }
+ if remote := uint64(db.Replica.Pos().TXID); remote != second {
+ t.Fatalf("the replica is at %d, the replicator reports %d", remote, second)
+ }
+}