| 1 | //go:build unix |
| 2 | |
| 3 | package host |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "slices" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/genselect" |
| 14 | "github.com/carlosframework/platform/internal/hostbox" |
| 15 | "github.com/carlosframework/platform/internal/hostclaim" |
| 16 | "github.com/carlosframework/platform/internal/hostguard" |
| 17 | "github.com/carlosframework/platform/internal/hostrelocate" |
| 18 | ) |
| 19 | |
| 20 | // probeRemote is the probe session's database as the region's bucket sees |
| 21 | // it: acked is the newest transaction the session acknowledged to a |
| 22 | // client, shipped the newest the replicator has put in the bucket. |
| 23 | // SyncAndWait ships everything acknowledged; once root's stop closes the |
| 24 | // replicator (closed) nothing more ships, and an unshipped write is gone |
| 25 | // with the session. A write is acknowledged only while the writers run. |
| 26 | type probeRemote struct { |
| 27 | acked, shipped uint64 |
| 28 | stopped, closed bool |
| 29 | syncErr error |
| 30 | } |
| 31 | |
| 32 | func (r *probeRemote) ack(t *testing.T) uint64 { |
| 33 | t.Helper() |
| 34 | if r.stopped || r.closed { |
| 35 | t.Fatal("a write acknowledged after the writers stopped") |
| 36 | } |
| 37 | r.acked++ |
| 38 | return r.acked |
| 39 | } |
| 40 | |
| 41 | func (r *probeRemote) SyncAndWait(context.Context) error { |
| 42 | if r.syncErr != nil { |
| 43 | return r.syncErr |
| 44 | } |
| 45 | if r.closed { |
| 46 | return errors.New("the replicator is closed") |
| 47 | } |
| 48 | r.shipped = r.acked |
| 49 | return nil |
| 50 | } |
| 51 | |
| 52 | // plan is the bucket's listing: what a cold restore reads. |
| 53 | func (r *probeRemote) plan(context.Context, genselect.Request) (*genselect.Plan, error) { |
| 54 | return &genselect.Plan{MaxTXID: r.shipped, MaxLevel0TXID: r.shipped}, nil |
| 55 | } |
| 56 | |
| 57 | // probeFlush is the probe stop as RelocationEffects.EndProbe runs it on the |
| 58 | // stop-probe fixture's target "dst" (the real relocationStopProbe, its |
| 59 | // activator and root's writer-session stop), with the release's own step |
| 60 | // providers: WritersStopped stops the session's writers, FlushDB is the |
| 61 | // release's releaseFinalTXID over the session's replicator, FlushFiles a |
| 62 | // recorded stand-in, Positions the real positions file. Root's stop |
| 63 | // closes the replicator. Every step lands in f.order. |
| 64 | type probeFlush struct { |
| 65 | f *stopProbeFixture |
| 66 | e *RelocationEffects |
| 67 | remote *probeRemote |
| 68 | positions ReleasePositionsFile |
| 69 | filesErr error |
| 70 | recordErr error |
| 71 | } |
| 72 | |
| 73 | func newProbeFlush(t *testing.T, passed time.Time) *probeFlush { |
| 74 | t.Helper() |
| 75 | f := newStopProbeFixture(t) |
| 76 | probe, _ := json.Marshal(hostclaim.Probe{Box: "dst", Session: 12, At: passed}) |
| 77 | f.obs.Claim.Op.Journal["probe"] = string(probe) |
| 78 | p := &probeFlush{f: f, remote: &probeRemote{acked: 40, shipped: 40}, positions: ReleasePositionsFile{Dir: t.TempDir()}} |
| 79 | stop := f.a.Wake.Stop |
| 80 | f.a.Wake.Stop = func(ctx context.Context, o hostclaim.Observed) error { |
| 81 | p.remote.closed = true // root's stop ends the session and its replicator |
| 82 | return stop(ctx, o) |
| 83 | } |
| 84 | p.e = &RelocationEffects{ |
| 85 | Identity: func(ctx context.Context) (RelocationIdentity, error) { return f.a.localStopIdentity(ctx) }, |
| 86 | WritersStopped: func(_ context.Context, o hostclaim.Observed) (ClaimWritersDrained, WritersStoppedEvidence, error) { |
| 87 | f.order = append(f.order, "writers-stopped") |
| 88 | p.remote.stopped = true |
| 89 | return probeWritersStopped(o) |
| 90 | }, |
| 91 | FlushDB: func(ctx context.Context, o hostclaim.Observed) (uint64, error) { |
| 92 | f.order = append(f.order, "db") |
| 93 | return releaseFinalTXID(ctx, p.remote, p.remote.plan, genselect.Request{Host: o.Host, Holder: *o.Claim.Holder}) |
| 94 | }, |
| 95 | FlushFiles: func(context.Context, hostclaim.Observed) (int64, error) { |
| 96 | f.order = append(f.order, "files") |
| 97 | return 3, p.filesErr |
| 98 | }, |
| 99 | Positions: probeFlushPositions{p}, |
| 100 | StopProbe: f.a.relocationStopProbe(f.cfg), |
| 101 | } |
| 102 | return p |
| 103 | } |
| 104 | |
| 105 | // probeFlushPositions is the real positions file, its writes in f.order. |
| 106 | type probeFlushPositions struct{ p *probeFlush } |
| 107 | |
| 108 | func (s probeFlushPositions) Record(ctx context.Context, r ReleasePosition) error { |
| 109 | s.p.f.order = append(s.p.f.order, "record") |
| 110 | if s.p.recordErr != nil { |
| 111 | return s.p.recordErr |
| 112 | } |
| 113 | return s.p.positions.Record(ctx, r) |
| 114 | } |
| 115 | |
| 116 | func (s probeFlushPositions) Recorded(ctx context.Context, r StopRequest) (ReleasePosition, bool, error) { |
| 117 | return s.p.positions.Recorded(ctx, r) |
| 118 | } |
| 119 | |
| 120 | // probeWritersStopped is a complete StopWriters witness and root |
| 121 | // observation for o's writer session on "dst". |
| 122 | func probeWritersStopped(o hostclaim.Observed) (ClaimWritersDrained, WritersStoppedEvidence, error) { |
| 123 | h := o.Claim.Holder |
| 124 | at := time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC) |
| 125 | target := hostguard.FenceTarget{Host: o.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Box: "dst", Enrollment: "enr", BootID: "boot", ControllerID: "ctl"} |
| 126 | drained := ClaimWritersDrained{Target: target, Generation: h.Generation, WriterBox: h.WriterBox, FenceClosed: true, LaunchJoined: true, WorkDrained: true, SpawnsSettled: true, |
| 127 | CgroupEmpty: true, Cgroup: "carlos/shop", DBPaths: []string{"/data/shop.db"}, CheckedAt: at} |
| 128 | seen := WritersStoppedEvidence{Host: o.Host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, WriterSession: h.Session, Box: "dst", Enrollment: "enr", |
| 129 | BootID: "boot", ControllerID: "ctl", ClaimETag: o.ETag, CheckedAt: at} |
| 130 | return drained, seen, nil |
| 131 | } |
| 132 | |
| 133 | func (p *probeFlush) end(t *testing.T) (hostclaim.ProbeStop, error) { |
| 134 | t.Helper() |
| 135 | return p.e.EndProbe(t.Context(), p.f.obs) |
| 136 | } |
| 137 | |
| 138 | // firstServingWake is the serving CAS then main's cold serving admission |
| 139 | // on "dst", at the bucket's MaxTXID as a cold restore reads it: its |
| 140 | // writes_lost declares the probe session's writes after that position. |
| 141 | func (p *probeFlush) firstServingWake(t *testing.T) hostclaim.AdmissionRecord { |
| 142 | t.Helper() |
| 143 | c := *p.f.obs.Claim |
| 144 | h := *c.Holder |
| 145 | c.Holder = &h |
| 146 | c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "dst", Participants: []string{"dst"}} |
| 147 | plan, err := p.remote.plan(t.Context(), genselect.Request{}) |
| 148 | if err != nil { |
| 149 | t.Fatal(err) |
| 150 | } |
| 151 | if err := hostclaim.AdmitWriter(hostclaim.Admission{ObservedSession: 12, ObservedGeneration: "P1", Box: "dst", Mode: "cold", NewGeneration: "S1", Region: "ie", RemoteMaxTXID: plan.MaxTXID})(&c); err != nil { |
| 152 | t.Fatal(err) |
| 153 | } |
| 154 | return c.Holder.Admissions[len(c.Holder.Admissions)-1] |
| 155 | } |
| 156 | |
| 157 | // Ruling 87: a write the probe session acknowledged just before its stop |
| 158 | // survives the first serving wake. The probe stop stops the writers, |
| 159 | // flushes the database (the release's releaseFinalTXID: the replicator |
| 160 | // ships every acknowledged write), flushes the files and records the |
| 161 | // positions, and only then fences and asks root's stop, which closes the |
| 162 | // replicator. The first serving wake is cold, as main's is, and its |
| 163 | // writes_lost names the probe session at the bucket's MaxTXID, which is |
| 164 | // now the acknowledged write's own: it declares nothing acknowledged lost. |
| 165 | func TestProbeStopFlushesSoAnAcknowledgedWriteSurvivesTheFirstServingWake(t *testing.T) { |
| 166 | p := newProbeFlush(t, time.Now()) |
| 167 | acked := p.remote.ack(t) // acknowledged; the replicator has not shipped it |
| 168 | if p.remote.shipped >= acked { |
| 169 | t.Fatal("the write shipped before the stop") |
| 170 | } |
| 171 | s, err := p.end(t) |
| 172 | if err != nil { |
| 173 | t.Fatal(err) |
| 174 | } |
| 175 | if want := []string{"writers-stopped", "db", "files", "record", "fence-as", "wake-stop"}; !slices.Equal(p.f.order, want) { |
| 176 | t.Fatalf("order %v, want %v", p.f.order, want) |
| 177 | } |
| 178 | if s.Box != "dst" || s.Session != 12 || s.Receipt != "fence-1" { |
| 179 | t.Fatalf("stop %+v", s) |
| 180 | } |
| 181 | req := probeStopRequest(p.f.obs, mustIdentity(t, p.f)) |
| 182 | rec, ok, err := p.positions.Recorded(t.Context(), req) |
| 183 | if err != nil || !ok || rec.TXID != acked || rec.Seq != 3 || rec.Request.ObligationID != hostguard.WriterStopObligation(12) || rec.Request.Phase != hostclaim.PhaseRelocatingProbing { |
| 184 | t.Fatalf("recorded %+v %v %v", rec, ok, err) |
| 185 | } |
| 186 | a := p.firstServingWake(t) |
| 187 | if a.Mode != "cold" || a.WritesLost == nil || a.WritesLost.Session != 12 || a.WritesLost.RemoteMaxTXID < acked { |
| 188 | t.Fatalf("the first serving wake declares %+v, the probe acknowledged %d", a.WritesLost, acked) |
| 189 | } |
| 190 | // A repeat (the probe-stop CAS lost) does not flush again: the |
| 191 | // replicator is closed, and the recorded positions are the flush. |
| 192 | p.f.order = nil |
| 193 | again, err := p.end(t) |
| 194 | if err != nil || again != s { |
| 195 | t.Fatalf("repeat: %+v %v", again, err) |
| 196 | } |
| 197 | if want := []string{"fence-as", "wake-stop"}; !slices.Equal(p.f.order, want) { |
| 198 | t.Fatalf("repeat order %v, want %v", p.f.order, want) |
| 199 | } |
| 200 | } |
| 201 | |
| 202 | func mustIdentity(t *testing.T, f *stopProbeFixture) RelocationIdentity { |
| 203 | t.Helper() |
| 204 | id, err := f.a.localStopIdentity(t.Context()) |
| 205 | if err != nil { |
| 206 | t.Fatal(err) |
| 207 | } |
| 208 | return id |
| 209 | } |
| 210 | |
| 211 | // Ruling 87's guards. A failed database flush, a failed files flush and a |
| 212 | // failed record each refuse the stop: nothing recorded, no fence, root |
| 213 | // not asked; the next pass flushes again and completes. Past the stop's |
| 214 | // deadline (ruling 67: ReleaseDeadline after the probe passed) the flush |
| 215 | // refuses before the writers are touched, yet a flush already recorded |
| 216 | // still lets the stop finish. Another session than the passed probe's is |
| 217 | // never flushed or stopped. |
| 218 | func TestProbeStopRefusesWithoutAFlush(t *testing.T) { |
| 219 | failure := errors.New("flush failed") |
| 220 | for name, fail := range map[string]func(*probeFlush){ |
| 221 | "the database": func(p *probeFlush) { p.remote.syncErr = failure }, |
| 222 | "the files": func(p *probeFlush) { p.filesErr = failure }, |
| 223 | "the record": func(p *probeFlush) { p.recordErr = failure }, |
| 224 | } { |
| 225 | t.Run(name, func(t *testing.T) { |
| 226 | p := newProbeFlush(t, time.Now()) |
| 227 | acked := p.remote.ack(t) |
| 228 | fail(p) |
| 229 | if _, err := p.end(t); !errors.Is(err, failure) { |
| 230 | t.Fatalf("a failed flush of %s: %v", name, err) |
| 231 | } |
| 232 | 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") { |
| 233 | t.Fatalf("the stop went ahead: %v", p.f.order) |
| 234 | } |
| 235 | if _, ok, err := p.positions.Recorded(t.Context(), probeStopRequest(p.f.obs, mustIdentity(t, p.f))); ok || err != nil { |
| 236 | t.Fatalf("a failed flush was recorded: %v", err) |
| 237 | } |
| 238 | // Retried once the flush works. |
| 239 | p.remote.syncErr, p.filesErr, p.recordErr = nil, nil, nil |
| 240 | p.f.order = nil |
| 241 | if _, err := p.end(t); err != nil { |
| 242 | t.Fatal(err) |
| 243 | } |
| 244 | if want := []string{"writers-stopped", "db", "files", "record", "fence-as", "wake-stop"}; !slices.Equal(p.f.order, want) || p.remote.shipped != acked { |
| 245 | t.Fatalf("retry order %v shipped %d", p.f.order, p.remote.shipped) |
| 246 | } |
| 247 | }) |
| 248 | } |
| 249 | t.Run("past the stop's deadline", func(t *testing.T) { |
| 250 | p := newProbeFlush(t, time.Now().Add(-hostrelocate.ReleaseDeadline-time.Minute)) |
| 251 | if _, err := p.end(t); !errors.Is(err, context.DeadlineExceeded) || len(p.f.order) != 0 { |
| 252 | t.Fatalf("past the deadline: %v %v", err, p.f.order) |
| 253 | } |
| 254 | }) |
| 255 | t.Run("past the deadline, already flushed", func(t *testing.T) { |
| 256 | p := newProbeFlush(t, time.Now()) |
| 257 | id := mustIdentity(t, p.f) |
| 258 | 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 { |
| 259 | t.Fatal(err) |
| 260 | } |
| 261 | probe, _ := json.Marshal(hostclaim.Probe{Box: "dst", Session: 12, At: time.Now().Add(-hostrelocate.ReleaseDeadline - time.Minute)}) |
| 262 | p.f.obs.Claim.Op.Journal["probe"] = string(probe) |
| 263 | if _, err := p.end(t); err != nil { |
| 264 | t.Fatal(err) |
| 265 | } |
| 266 | if want := []string{"fence-as", "wake-stop"}; !slices.Equal(p.f.order, want) { |
| 267 | t.Fatalf("order %v, want %v", p.f.order, want) |
| 268 | } |
| 269 | }) |
| 270 | t.Run("not the probe's session", func(t *testing.T) { |
| 271 | p := newProbeFlush(t, time.Now()) |
| 272 | p.f.obs.Claim.Holder.Session = 13 |
| 273 | if _, err := p.end(t); !errors.Is(err, hostbox.ErrChanged) || len(p.f.order) != 0 || p.remote.stopped { |
| 274 | t.Fatalf("another session: %v %v", err, p.f.order) |
| 275 | } |
| 276 | }) |
| 277 | t.Run("an unreadable probe", func(t *testing.T) { |
| 278 | p := newProbeFlush(t, time.Now()) |
| 279 | p.f.obs.Claim.Op.Journal["probe"] = "{" |
| 280 | if _, err := p.end(t); err == nil || len(p.f.order) != 0 { |
| 281 | t.Fatalf("an unreadable probe: %v %v", err, p.f.order) |
| 282 | } |
| 283 | }) |
| 284 | } |
| 285 | |
amadan