| new file mode 100644 |
| index 000000000..c375cac0f |
| --- /dev/null |
| +++ b/internal/host/relocationprobeflush_unix_test.go |
| @@ -0,0 +1,284 @@ |
| +//go:build unix |
| + |
| +package host |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "slices" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/genselect" |
| + "github.com/carlosframework/platform/internal/hostbox" |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/hostguard" |
| + "github.com/carlosframework/platform/internal/hostrelocate" |
| +) |
| + |
| +// probeRemote is the probe session's database as the region's bucket sees |
| +// it: acked is the newest transaction the session acknowledged to a |
| +// client, shipped the newest the replicator has put in the bucket. |
| +// SyncAndWait ships everything acknowledged; once root's stop closes the |
| +// replicator (closed) nothing more ships, and an unshipped write is gone |
| +// with the session. A write is acknowledged only while the writers run. |
| +type probeRemote struct { |
| + acked, shipped uint64 |
| + stopped, closed bool |
| + syncErr error |
| +} |
| + |
| +func (r *probeRemote) ack(t *testing.T) uint64 { |
| + t.Helper() |
| + if r.stopped || r.closed { |
| + t.Fatal("a write acknowledged after the writers stopped") |
| + } |
| + r.acked++ |
| + return r.acked |
| +} |
| + |
| +func (r *probeRemote) SyncAndWait(context.Context) error { |
| + if r.syncErr != nil { |
| + return r.syncErr |
| + } |
| + if r.closed { |
| + return errors.New("the replicator is closed") |
| + } |
| + r.shipped = r.acked |
| + return nil |
| +} |
| + |
| +// plan is the bucket's listing: what a cold restore reads. |
| +func (r *probeRemote) plan(context.Context, genselect.Request) (*genselect.Plan, error) { |
| + return &genselect.Plan{MaxTXID: r.shipped, MaxLevel0TXID: r.shipped}, nil |
| +} |
| + |
| +// probeFlush is the probe stop as RelocationEffects.EndProbe runs it on the |
| +// stop-probe fixture's target "dst" (the real relocationStopProbe, its |
| +// activator and root's writer-session stop), with the release's own step |
| +// providers: WritersStopped stops the session's writers, FlushDB is the |
| +// release's releaseFinalTXID over the session's replicator, FlushFiles a |
| +// recorded stand-in, Positions the real positions file. Root's stop |
| +// closes the replicator. Every step lands in f.order. |
| +type probeFlush struct { |
| + f *stopProbeFixture |
| + e *RelocationEffects |
| + remote *probeRemote |
| + positions ReleasePositionsFile |
| + filesErr error |
| + recordErr error |
| +} |
| + |
| +func newProbeFlush(t *testing.T, passed time.Time) *probeFlush { |
| + t.Helper() |
| + f := newStopProbeFixture(t) |
| + probe, _ := json.Marshal(hostclaim.Probe{Box: "dst", Session: 12, At: passed}) |
| + f.obs.Claim.Op.Journal["probe"] = string(probe) |
| + p := &probeFlush{f: f, remote: &probeRemote{acked: 40, shipped: 40}, positions: ReleasePositionsFile{Dir: t.TempDir()}} |
| + stop := f.a.Wake.Stop |
| + f.a.Wake.Stop = func(ctx context.Context, o hostclaim.Observed) error { |
| + p.remote.closed = true // root's stop ends the session and its replicator |
| + return stop(ctx, o) |
| + } |
| + p.e = &RelocationEffects{ |
| + Identity: func(ctx context.Context) (RelocationIdentity, error) { return f.a.localStopIdentity(ctx) }, |
| + WritersStopped: func(_ context.Context, o hostclaim.Observed) (ClaimWritersDrained, WritersStoppedEvidence, error) { |
| + f.order = append(f.order, "writers-stopped") |
| + p.remote.stopped = true |
| + return probeWritersStopped(o) |
| + }, |
| + FlushDB: func(ctx context.Context, o hostclaim.Observed) (uint64, error) { |
| + f.order = append(f.order, "db") |
| + return releaseFinalTXID(ctx, p.remote, p.remote.plan, genselect.Request{Host: o.Host, Holder: *o.Claim.Holder}) |
| + }, |
| + FlushFiles: func(context.Context, hostclaim.Observed) (int64, error) { |
| + f.order = append(f.order, "files") |
| + return 3, p.filesErr |
| + }, |
| + Positions: probeFlushPositions{p}, |
| + StopProbe: f.a.relocationStopProbe(f.cfg), |
| + } |
| + return p |
| +} |
| + |
| +// probeFlushPositions is the real positions file, its writes in f.order. |
| +type probeFlushPositions struct{ p *probeFlush } |
| + |
| +func (s probeFlushPositions) Record(ctx context.Context, r ReleasePosition) error { |
| + s.p.f.order = append(s.p.f.order, "record") |
| + if s.p.recordErr != nil { |
| + return s.p.recordErr |
| + } |
| + return s.p.positions.Record(ctx, r) |
| +} |
| + |
| +func (s probeFlushPositions) Recorded(ctx context.Context, r StopRequest) (ReleasePosition, bool, error) { |
| + return s.p.positions.Recorded(ctx, r) |
| +} |
| + |
| +// probeWritersStopped is a complete StopWriters witness and root |
| +// observation for o's writer session on "dst". |
| +func probeWritersStopped(o hostclaim.Observed) (ClaimWritersDrained, WritersStoppedEvidence, error) { |
| + h := o.Claim.Holder |
| + at := time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC) |
| + target := hostguard.FenceTarget{Host: o.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Box: "dst", Enrollment: "enr", BootID: "boot", ControllerID: "ctl"} |
| + drained := ClaimWritersDrained{Target: target, Generation: h.Generation, WriterBox: h.WriterBox, FenceClosed: true, LaunchJoined: true, WorkDrained: true, SpawnsSettled: true, |
| + CgroupEmpty: true, Cgroup: "carlos/shop", DBPaths: []string{"/data/shop.db"}, CheckedAt: at} |
| + seen := WritersStoppedEvidence{Host: o.Host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, WriterSession: h.Session, Box: "dst", Enrollment: "enr", |
| + BootID: "boot", ControllerID: "ctl", ClaimETag: o.ETag, CheckedAt: at} |
| + return drained, seen, nil |
| +} |
| + |
| +func (p *probeFlush) end(t *testing.T) (hostclaim.ProbeStop, error) { |
| + t.Helper() |
| + return p.e.EndProbe(t.Context(), p.f.obs) |
| +} |
| + |
| +// firstServingWake is the serving CAS then main's cold serving admission |
| +// on "dst", at the bucket's MaxTXID as a cold restore reads it: its |
| +// writes_lost declares the probe session's writes after that position. |
| +func (p *probeFlush) firstServingWake(t *testing.T) hostclaim.AdmissionRecord { |
| + t.Helper() |
| + c := *p.f.obs.Claim |
| + h := *c.Holder |
| + c.Holder = &h |
| + c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "dst", Participants: []string{"dst"}} |
| + plan, err := p.remote.plan(t.Context(), genselect.Request{}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := hostclaim.AdmitWriter(hostclaim.Admission{ObservedSession: 12, ObservedGeneration: "P1", Box: "dst", Mode: "cold", NewGeneration: "S1", Region: "ie", RemoteMaxTXID: plan.MaxTXID})(&c); err != nil { |
| + t.Fatal(err) |
| + } |
| + return c.Holder.Admissions[len(c.Holder.Admissions)-1] |
| +} |
| + |
| +// Ruling 87: a write the probe session acknowledged just before its stop |
| +// survives the first serving wake. The probe stop stops the writers, |
| +// flushes the database (the release's releaseFinalTXID: the replicator |
| +// ships every acknowledged write), flushes the files and records the |
| +// positions, and only then fences and asks root's stop, which closes the |
| +// replicator. The first serving wake is cold, as main's is, and its |
| +// writes_lost names the probe session at the bucket's MaxTXID, which is |
| +// now the acknowledged write's own: it declares nothing acknowledged lost. |
| +func TestProbeStopFlushesSoAnAcknowledgedWriteSurvivesTheFirstServingWake(t *testing.T) { |
| + p := newProbeFlush(t, time.Now()) |
| + acked := p.remote.ack(t) // acknowledged; the replicator has not shipped it |
| + if p.remote.shipped >= acked { |
| + t.Fatal("the write shipped before the stop") |
| + } |
| + s, err := p.end(t) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if want := []string{"writers-stopped", "db", "files", "record", "fence-as", "wake-stop"}; !slices.Equal(p.f.order, want) { |
| + t.Fatalf("order %v, want %v", p.f.order, want) |
| + } |
| + if s.Box != "dst" || s.Session != 12 || s.Receipt != "fence-1" { |
| + t.Fatalf("stop %+v", s) |
| + } |
| + req := probeStopRequest(p.f.obs, mustIdentity(t, p.f)) |
| + rec, ok, err := p.positions.Recorded(t.Context(), req) |
| + if err != nil || !ok || rec.TXID != acked || rec.Seq != 3 || rec.Request.ObligationID != hostguard.WriterStopObligation(12) || rec.Request.Phase != hostclaim.PhaseRelocatingProbing { |
| + t.Fatalf("recorded %+v %v %v", rec, ok, err) |
| + } |
| + a := p.firstServingWake(t) |
| + if a.Mode != "cold" || a.WritesLost == nil || a.WritesLost.Session != 12 || a.WritesLost.RemoteMaxTXID < acked { |
| + t.Fatalf("the first serving wake declares %+v, the probe acknowledged %d", a.WritesLost, acked) |
| + } |
| + // A repeat (the probe-stop CAS lost) does not flush again: the |
| + // replicator is closed, and the recorded positions are the flush. |
| + p.f.order = nil |
| + again, err := p.end(t) |
| + if err != nil || again != s { |
| + t.Fatalf("repeat: %+v %v", again, err) |
| + } |
| + if want := []string{"fence-as", "wake-stop"}; !slices.Equal(p.f.order, want) { |
| + t.Fatalf("repeat order %v, want %v", p.f.order, want) |
| + } |
| +} |
| + |
| +func mustIdentity(t *testing.T, f *stopProbeFixture) RelocationIdentity { |
| + t.Helper() |
| + id, err := f.a.localStopIdentity(t.Context()) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return id |
| +} |
| + |
| +// Ruling 87's guards. A failed database flush, a failed files flush and a |
| +// failed record each refuse the stop: nothing recorded, no fence, root |
| +// not asked; the next pass flushes again and completes. Past the stop's |
| +// deadline (ruling 67: ReleaseDeadline after the probe passed) the flush |
| +// refuses before the writers are touched, yet a flush already recorded |
| +// still lets the stop finish. Another session than the passed probe's is |
| +// never flushed or stopped. |
| +func TestProbeStopRefusesWithoutAFlush(t *testing.T) { |
| + failure := errors.New("flush failed") |
| + for name, fail := range map[string]func(*probeFlush){ |
| + "the database": func(p *probeFlush) { p.remote.syncErr = failure }, |
| + "the files": func(p *probeFlush) { p.filesErr = failure }, |
| + "the record": func(p *probeFlush) { p.recordErr = failure }, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + p := newProbeFlush(t, time.Now()) |
| + acked := p.remote.ack(t) |
| + fail(p) |
| + if _, err := p.end(t); !errors.Is(err, failure) { |
| + t.Fatalf("a failed flush of %s: %v", name, err) |
| + } |
| + if slices.Contains(p.f.order, "fence-as") || slices.Contains(p.f.order, "wake-stop") || p.remote.closed || p.f.act.svc.Fenced("shop.example") { |
| + t.Fatalf("the stop went ahead: %v", p.f.order) |
| + } |
| + if _, ok, err := p.positions.Recorded(t.Context(), probeStopRequest(p.f.obs, mustIdentity(t, p.f))); ok || err != nil { |
| + t.Fatalf("a failed flush was recorded: %v", err) |
| + } |
| + // Retried once the flush works. |
| + p.remote.syncErr, p.filesErr, p.recordErr = nil, nil, nil |
| + p.f.order = nil |
| + if _, err := p.end(t); err != nil { |
| + t.Fatal(err) |
| + } |
| + if want := []string{"writers-stopped", "db", "files", "record", "fence-as", "wake-stop"}; !slices.Equal(p.f.order, want) || p.remote.shipped != acked { |
| + t.Fatalf("retry order %v shipped %d", p.f.order, p.remote.shipped) |
| + } |
| + }) |
| + } |
| + t.Run("past the stop's deadline", func(t *testing.T) { |
| + p := newProbeFlush(t, time.Now().Add(-hostrelocate.ReleaseDeadline-time.Minute)) |
| + if _, err := p.end(t); !errors.Is(err, context.DeadlineExceeded) || len(p.f.order) != 0 { |
| + t.Fatalf("past the deadline: %v %v", err, p.f.order) |
| + } |
| + }) |
| + t.Run("past the deadline, already flushed", func(t *testing.T) { |
| + p := newProbeFlush(t, time.Now()) |
| + id := mustIdentity(t, p.f) |
| + if err := p.positions.Record(t.Context(), ReleasePosition{Request: probeStopRequest(p.f.obs, id), PID: id.PID, ProcessStart: id.ProcessStart, TXID: 40, Seq: 3}); err != nil { |
| + t.Fatal(err) |
| + } |
| + probe, _ := json.Marshal(hostclaim.Probe{Box: "dst", Session: 12, At: time.Now().Add(-hostrelocate.ReleaseDeadline - time.Minute)}) |
| + p.f.obs.Claim.Op.Journal["probe"] = string(probe) |
| + if _, err := p.end(t); err != nil { |
| + t.Fatal(err) |
| + } |
| + if want := []string{"fence-as", "wake-stop"}; !slices.Equal(p.f.order, want) { |
| + t.Fatalf("order %v, want %v", p.f.order, want) |
| + } |
| + }) |
| + t.Run("not the probe's session", func(t *testing.T) { |
| + p := newProbeFlush(t, time.Now()) |
| + p.f.obs.Claim.Holder.Session = 13 |
| + if _, err := p.end(t); !errors.Is(err, hostbox.ErrChanged) || len(p.f.order) != 0 || p.remote.stopped { |
| + t.Fatalf("another session: %v %v", err, p.f.order) |
| + } |
| + }) |
| + t.Run("an unreadable probe", func(t *testing.T) { |
| + p := newProbeFlush(t, time.Now()) |
| + p.f.obs.Claim.Op.Journal["probe"] = "{" |
| + if _, err := p.end(t); err == nil || len(p.f.order) != 0 { |
| + t.Fatalf("an unreadable probe: %v %v", err, p.f.order) |
| + } |
| + }) |
| +} |
amadan