| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "strings" |
| 11 | "sync" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/hostbox" |
| 16 | "github.com/carlosframework/platform/internal/hostclaim" |
| 17 | "github.com/carlosframework/platform/internal/hostguard" |
| 18 | "github.com/carlosframework/platform/internal/hostwrite" |
| 19 | "github.com/carlosframework/platform/internal/moves" |
| 20 | "github.com/carlosframework/platform/internal/provision" |
| 21 | "github.com/carlosframework/platform/internal/registry" |
| 22 | "github.com/carlosframework/platform/internal/release" |
| 23 | "github.com/carlosframework/platform/internal/rootstate" |
| 24 | ) |
| 25 | |
| 26 | // Task 7b: the root-side held release under the host claim (moves after |
| 27 | // claim Plan 12; task-7b-design.md and platform-1e's binding review of |
| 28 | // 2026-10-12). The sandboxed agent cannot write /opt/carlos, so under the |
| 29 | // claim it asks root, over the root control socket, to stage T7's release |
| 30 | // (never publishing it) and to publish the held row's release at configured |
| 31 | // and before a probe's start. Root re-reads the claim, the record, the |
| 32 | // registry row and the render-hash journal, refuses anything that does not |
| 33 | // match before any effect, and publishes only inside its own held-release |
| 34 | // selection. |
| 35 | |
| 36 | const ( |
| 37 | hrRendered = "sha256:" + "aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11aa11" |
| 38 | hrOther = "sha256:" + "bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22bb22" |
| 39 | hrRB = "rb_abcdefghijklmnopqrstuvwxyz" |
| 40 | ) |
| 41 | |
| 42 | var hrPeer = StopPeer{UID: 1000, PID: 4242, ProcessStart: "77", BootID: "boot", ControllerID: "edge"} |
| 43 | |
| 44 | // hrSelector is root's held-release selection as the tests see it: each |
| 45 | // selection is recorded with the claim tuple it was made at, and each write |
| 46 | // it is handed runs, recording the identity it names. |
| 47 | type hrSelector struct { |
| 48 | mu sync.Mutex |
| 49 | f *hrFixture |
| 50 | selected []string |
| 51 | ids []hostwrite.Identity |
| 52 | refuse error |
| 53 | } |
| 54 | |
| 55 | type hrWork struct { |
| 56 | s *hrSelector |
| 57 | sel hostbox.Selection |
| 58 | } |
| 59 | |
| 60 | func (w hrWork) With(_ context.Context, id hostwrite.Identity, fn func() error) error { |
| 61 | w.s.mu.Lock() |
| 62 | w.s.ids = append(w.s.ids, id) |
| 63 | w.s.mu.Unlock() |
| 64 | return fn() |
| 65 | } |
| 66 | |
| 67 | func (s *hrSelector) count() int { |
| 68 | s.mu.Lock() |
| 69 | defer s.mu.Unlock() |
| 70 | return len(s.selected) |
| 71 | } |
| 72 | |
| 73 | // hrFixture is one box (uk-box in uk) with a held row of bab/app at a |
| 74 | // landing's configured step: the claim, the record, the registry row, the |
| 75 | // render journal, a store with v7 shipped and authorized, and root. |
| 76 | type hrFixture struct { |
| 77 | s release.Store |
| 78 | opt, data, receipts string |
| 79 | inc string |
| 80 | mu sync.Mutex |
| 81 | claim hostclaim.Claim |
| 82 | rec provision.Record |
| 83 | row *registry.Route |
| 84 | sel *hrSelector |
| 85 | claimReads int |
| 86 | root *movesHeldReleaseRoot |
| 87 | } |
| 88 | |
| 89 | func newHRFixture(t *testing.T) *hrFixture { |
| 90 | t.Helper() |
| 91 | s := release.NewDirStore(t.TempDir()) |
| 92 | shipPinnedV7(t, s) |
| 93 | inc, err := hostclaim.Incarnation("bab", "app", p1Host, p1Created) |
| 94 | if err != nil { |
| 95 | t.Fatal(err) |
| 96 | } |
| 97 | f := &hrFixture{s: s, opt: t.TempDir(), data: t.TempDir(), receipts: t.TempDir(), inc: inc} |
| 98 | f.claim = hostclaim.Claim{V: 1, Host: p1Host, Gen: 5, |
| 99 | Holder: &hostclaim.Holder{Family: "instance", Key: provision.RecordKey("bab", "app", p1Host), Account: "bab", App: "app", Incarnation: inc, |
| 100 | CreatedAt: p1Created, Location: "instances/" + p1Host + "/" + inc, Generation: p1ID, Stamp: 7, Session: 3, WriterBox: "uk-box"}, |
| 101 | Op: hostclaim.Op{Kind: hostclaim.KindLanding, ID: p1ID, Phase: provision.LandConfigured, Exec: "uk-box", Participants: []string{"ie-box", "uk-box"}, |
| 102 | Intent: map[string]string{"source": "ie-box", "target": "uk-box", "to": "uk"}}} |
| 103 | f.rec = provision.Record{Schema: 1, Host: p1Host, Slug: "s", App: "app", Account: "bab", Region: "uk", CreatedAt: p1Created, |
| 104 | Move: &provision.MoveState{ID: p1ID, Incarnation: inc, From: "ie", To: "uk", State: provision.MoveLanding, LandingPhase: provision.LandConfigured, |
| 105 | Inputs: &provision.MoveInputs{Release: "v7"}}} |
| 106 | f.row = ®istry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", Status: "landing", HoldOp: p1ID} |
| 107 | f.sel = &hrSelector{f: f} |
| 108 | raw, err := json.Marshal(moves.SourceReport{V: 1, Box: "ie-box", MoveID: p1ID, At: time.Now(), Release: "v7"}) |
| 109 | if err != nil { |
| 110 | t.Fatal(err) |
| 111 | } |
| 112 | if _, err := s.Put(context.Background(), moves.SourceReportKey(moves.ReportPrefix("bab", "app", p1Host, p1ID)), raw, ""); err != nil { |
| 113 | t.Fatal(err) |
| 114 | } |
| 115 | f.journal(t, p1ID, hrRendered) |
| 116 | f.root = &movesHeldReleaseRoot{ |
| 117 | OpenRelease: func() (rootstate.Store, error) { return &plan14ReleaseStore{}, nil }, |
| 118 | Authenticate: nil, |
| 119 | Actor: func(context.Context) (hostbox.Actor, error) { |
| 120 | return hostbox.Actor{Box: "uk-box", Region: "uk", Eligible: true}, nil |
| 121 | }, |
| 122 | Store: s, |
| 123 | Claims: f.claims, |
| 124 | Records: f.records, |
| 125 | Route: func(host string) (*registry.Route, error) { |
| 126 | f.mu.Lock() |
| 127 | defer f.mu.Unlock() |
| 128 | if f.row == nil || host != f.row.Host { |
| 129 | return nil, nil |
| 130 | } |
| 131 | rt := *f.row |
| 132 | return &rt, nil |
| 133 | }, |
| 134 | Works: func(_ context.Context, actor hostbox.Actor, host, action string) (hostwrite.Work, error) { |
| 135 | f.sel.mu.Lock() |
| 136 | defer f.sel.mu.Unlock() |
| 137 | f.sel.selected = append(f.sel.selected, host+"/"+action) |
| 138 | if f.sel.refuse != nil { |
| 139 | return nil, f.sel.refuse |
| 140 | } |
| 141 | f.mu.Lock() |
| 142 | c := f.claim |
| 143 | f.mu.Unlock() |
| 144 | return hrWork{s: f.sel, sel: hostbox.Selection{Host: host, Action: action, Box: actor.Box, Region: actor.Region, |
| 145 | Captured: hostguard.Captured{Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp}, Kind: c.Op.Kind, Op: c.Op.ID, Phase: c.Op.Phase, |
| 146 | Generation: c.Holder.Generation, Session: c.Holder.Session}}, nil |
| 147 | }, |
| 148 | Selected: func(w hostwrite.Work) (hostbox.Selection, error) { |
| 149 | hw, ok := w.(hrWork) |
| 150 | if !ok { |
| 151 | return hostbox.Selection{}, hostbox.ErrUnsupported |
| 152 | } |
| 153 | return hw.sel, nil |
| 154 | }, |
| 155 | Pub: hsPub, |
| 156 | Ratchet: func() *release.DigestRatchet { return hsRatchet(t) }, |
| 157 | Proof: func() *release.ProofGate { return offProofGate(t) }, |
| 158 | OptDir: f.opt, |
| 159 | DataDir: f.data, |
| 160 | ReceiptDir: f.receipts, |
| 161 | } |
| 162 | return f |
| 163 | } |
| 164 | |
| 165 | func (f *hrFixture) claims(_ context.Context, host string) (hostclaim.Observed, error) { |
| 166 | f.mu.Lock() |
| 167 | defer f.mu.Unlock() |
| 168 | f.claimReads++ |
| 169 | if host != f.claim.Host { |
| 170 | return hostclaim.Observed{Host: host, Absent: true}, nil |
| 171 | } |
| 172 | raw, err := json.Marshal(f.claim) |
| 173 | if err != nil { |
| 174 | return hostclaim.Observed{}, err |
| 175 | } |
| 176 | var c hostclaim.Claim |
| 177 | if err := json.Unmarshal(raw, &c); err != nil { |
| 178 | return hostclaim.Observed{}, err |
| 179 | } |
| 180 | return hostclaim.Observed{Host: host, Claim: &c}, nil |
| 181 | } |
| 182 | |
| 183 | func (f *hrFixture) records(_ context.Context, acct, app, host string) (*provision.Record, error) { |
| 184 | f.mu.Lock() |
| 185 | defer f.mu.Unlock() |
| 186 | if acct != f.rec.Account || app != f.rec.App || host != f.rec.Host { |
| 187 | return nil, nil |
| 188 | } |
| 189 | raw, err := json.Marshal(f.rec) |
| 190 | if err != nil { |
| 191 | return nil, err |
| 192 | } |
| 193 | var rec provision.Record |
| 194 | if err := json.Unmarshal(raw, &rec); err != nil { |
| 195 | return nil, err |
| 196 | } |
| 197 | return &rec, nil |
| 198 | } |
| 199 | |
| 200 | // journal journals the held render's hash for op, as configured does. |
| 201 | func (f *hrFixture) journal(t *testing.T, op, hash string) { |
| 202 | t.Helper() |
| 203 | if err := moves.UpdateJournal(f.data, p1Host, op, func(j *moves.Journal) error { |
| 204 | j.Mark(landStepRendered, hash) |
| 205 | return nil |
| 206 | }); err != nil { |
| 207 | t.Fatal(err) |
| 208 | } |
| 209 | } |
| 210 | |
| 211 | // request is the agent's request for action, built from the fixture's claim |
| 212 | // and record as the agent builds it from its own reads. |
| 213 | func (f *hrFixture) request(action string) heldReleaseRequest { |
| 214 | f.mu.Lock() |
| 215 | defer f.mu.Unlock() |
| 216 | c := f.claim |
| 217 | q := heldReleaseRequest{Schema: heldReleaseSchema, RequestID: "req-1", Action: action, Host: p1Host, Account: "bab", App: "app", Version: "v7", |
| 218 | MoveID: c.Op.ID, HoldOp: c.Op.ID, Box: "uk-box", Region: "uk", Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp, |
| 219 | Session: c.Holder.Session, Generation: c.Holder.Generation, Kind: c.Op.Kind, Phase: c.Op.Phase} |
| 220 | if c.Op.Kind == hostclaim.KindRestoring { |
| 221 | q.HoldOp = c.Op.Intent["rollback"] |
| 222 | } |
| 223 | if action == heldReleasePublish { |
| 224 | q.RenderHash = hrRendered |
| 225 | } |
| 226 | return q |
| 227 | } |
| 228 | |
| 229 | func (f *hrFixture) set(fn func(f *hrFixture)) { |
| 230 | f.mu.Lock() |
| 231 | defer f.mu.Unlock() |
| 232 | fn(f) |
| 233 | } |
| 234 | |
| 235 | // untouched fails unless root had no effect: nothing selected, nothing |
| 236 | // staged, live/<host> absent and no receipt. |
| 237 | func (f *hrFixture) untouched(t *testing.T) { |
| 238 | t.Helper() |
| 239 | if n := f.sel.count(); n != 0 { |
| 240 | t.Fatalf("root selected %d time(s) for a refused request", n) |
| 241 | } |
| 242 | if _, err := os.Stat(filepath.Join(f.opt, "versions")); !errors.Is(err, os.ErrNotExist) { |
| 243 | t.Fatalf("root staged for a refused request: %v", err) |
| 244 | } |
| 245 | if _, err := os.Lstat(release.LiveLink(f.opt, p1Host)); !errors.Is(err, os.ErrNotExist) { |
| 246 | t.Fatalf("root published for a refused request: %v", err) |
| 247 | } |
| 248 | if ents, _ := os.ReadDir(f.receipts); len(ents) != 0 { |
| 249 | t.Fatalf("root wrote a receipt for a refused request: %v", ents) |
| 250 | } |
| 251 | } |
| 252 | |
| 253 | // TestHeldReleaseRootPublishesAtConfigured: a landing's configured step |
| 254 | // asks root to publish the pinned release; root selects held-release for |
| 255 | // the host, publishes live/<host> under the identity naming exactly v7, |
| 256 | // writes the receipt, and the agent's own reading then passes. |
| 257 | func TestHeldReleaseRootPublishesAtConfigured(t *testing.T) { |
| 258 | ctx := context.Background() |
| 259 | f := newHRFixture(t) |
| 260 | resp, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)) |
| 261 | if err != nil { |
| 262 | t.Fatalf("publish: %v", err) |
| 263 | } |
| 264 | if resp.Outcome != heldReleaseOutcomePublished || resp.Replayed || resp.RequestID != "req-1" { |
| 265 | t.Fatalf("response = %+v", resp) |
| 266 | } |
| 267 | if v, _ := release.LiveVersion(f.opt, p1Host); v != "v7" { |
| 268 | t.Fatalf("live = %q, want v7", v) |
| 269 | } |
| 270 | if len(f.sel.selected) != 1 || f.sel.selected[0] != p1Host+"/"+movesActionHeldRelease { |
| 271 | t.Fatalf("selected %v, want one held-release selection", f.sel.selected) |
| 272 | } |
| 273 | if len(f.sel.ids) != 1 || f.sel.ids[0] != (hostwrite.Identity{Release: "v7"}) { |
| 274 | t.Fatalf("published under %v, want the pinned release alone", f.sel.ids) |
| 275 | } |
| 276 | if _, err := os.Stat(filepath.Join(f.receipts, p1Host+".json")); err != nil { |
| 277 | t.Fatalf("no publish receipt: %v", err) |
| 278 | } |
| 279 | rt := registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", HoldOp: p1ID} |
| 280 | lock := func(read func() error) error { return withHeldStageLock(f.data, p1Host, read) } |
| 281 | if err := release.VerifyHeldLive(ctx, f.s, f.opt, rt, "v7", offProofGate(t), lock); err != nil { |
| 282 | t.Fatalf("the agent's reading after root's publication: %v", err) |
| 283 | } |
| 284 | } |
| 285 | |
| 286 | // TestHeldReleaseRootPublishesForTheDirtyReturnAndTheProbe: the same path |
| 287 | // serves a landing's probe and a dirty return, at its configured (claimed) |
| 288 | // and its probe, each under its own hold op and render journal. |
| 289 | func TestHeldReleaseRootPublishesForTheDirtyReturnAndTheProbe(t *testing.T) { |
| 290 | ctx := context.Background() |
| 291 | asReturn := func(f *hrFixture, phase, sourcePhase string, recovery *provision.MoveInputs) { |
| 292 | f.claim.Op = hostclaim.Op{Kind: hostclaim.KindRestoring, ID: p1ID, Phase: phase, Exec: "uk-box", Participants: []string{"uk-box", "ie-box"}, |
| 293 | Intent: map[string]string{"source": "uk-box", "target": "ie-box", "to": "ie", "rollback": hrRB}} |
| 294 | f.rec.Move.State, f.rec.Move.LandingPhase = provision.MoveRollingBack, "" |
| 295 | f.rec.Move.From, f.rec.Move.To = "uk", "ie" |
| 296 | f.rec.Move.Rollback = &provision.MoveRollback{ID: hrRB, Dirty: true, SourcePhase: sourcePhase, RecoveryInputs: recovery} |
| 297 | f.row.HoldOp = hrRB |
| 298 | } |
| 299 | cases := []struct { |
| 300 | name string |
| 301 | set func(f *hrFixture) |
| 302 | op string |
| 303 | }{ |
| 304 | {"the landing's probe", func(f *hrFixture) { |
| 305 | f.claim.Op.Phase = provision.LandProbing |
| 306 | f.rec.Move.LandingPhase = provision.LandProbing |
| 307 | }, p1ID}, |
| 308 | {"a dirty return's configured", func(f *hrFixture) { |
| 309 | asReturn(f, hostclaim.PhaseClaimed, provision.RbSourceConfigured, nil) |
| 310 | }, hrRB}, |
| 311 | {"a dirty return's probe", func(f *hrFixture) { |
| 312 | asReturn(f, hostclaim.PhaseProbing, provision.RbSourceProbing, &provision.MoveInputs{Release: "v7"}) |
| 313 | }, hrRB}, |
| 314 | } |
| 315 | for _, tc := range cases { |
| 316 | t.Run(tc.name, func(t *testing.T) { |
| 317 | f := newHRFixture(t) |
| 318 | f.set(tc.set) |
| 319 | f.journal(t, tc.op, hrRendered) |
| 320 | q := f.request(heldReleasePublish) |
| 321 | if q.HoldOp != tc.op { |
| 322 | t.Fatalf("request hold op %q, want %q", q.HoldOp, tc.op) |
| 323 | } |
| 324 | resp, err := f.root.Serve(ctx, hrPeer, q) |
| 325 | if err != nil || resp.Outcome != heldReleaseOutcomePublished { |
| 326 | t.Fatalf("publish: %+v %v", resp, err) |
| 327 | } |
| 328 | if v, _ := release.LiveVersion(f.opt, p1Host); v != "v7" { |
| 329 | t.Fatalf("live = %q, want v7", v) |
| 330 | } |
| 331 | if len(f.sel.ids) != 1 || f.sel.ids[0] != (hostwrite.Identity{Release: "v7"}) { |
| 332 | t.Fatalf("published under %v", f.sel.ids) |
| 333 | } |
| 334 | }) |
| 335 | } |
| 336 | } |
| 337 | |
| 338 | // TestHeldReleaseRootRefusesBeforeAnyEffect: every request whose tuple is |
| 339 | // not the claim's, whose pins or render are not the record's and the |
| 340 | // journal's, that names another account, app, box or region, or that |
| 341 | // arrives while the claim grants no held write, is refused before root |
| 342 | // selects, stages, publishes or writes a receipt. |
| 343 | func TestHeldReleaseRootRefusesBeforeAnyEffect(t *testing.T) { |
| 344 | t.Parallel() |
| 345 | ctx := context.Background() |
| 346 | otherInc := strings.Repeat("c", 64) |
| 347 | cases := []struct { |
| 348 | name string |
| 349 | action string |
| 350 | state func(f *hrFixture) |
| 351 | q func(q *heldReleaseRequest) |
| 352 | }{ |
| 353 | {"a stale incarnation", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Incarnation = otherInc }}, |
| 354 | {"a stale stamp", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Stamp = 6 }}, |
| 355 | {"a stale session", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Session = 2 }}, |
| 356 | {"a stale generation", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Generation = "cur" }}, |
| 357 | {"a stale op", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.MoveID, q.HoldOp = p1ID2, p1ID2 }}, |
| 358 | {"a stale phase", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Phase = provision.LandProbing }}, |
| 359 | {"a stale kind", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Kind = hostclaim.KindReverting }}, |
| 360 | {"changed pins: another version", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Version = "v8" }}, |
| 361 | {"changed pins: the record's", heldReleasePublish, func(f *hrFixture) { f.rec.Move.Inputs.Release = "v8" }, nil}, |
| 362 | {"changed render", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.RenderHash = hrOther }}, |
| 363 | {"a forged account", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Account = "zed" }}, |
| 364 | {"a forged app", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.App = "other" }}, |
| 365 | {"the wrong box", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Box = "ie-box" }}, |
| 366 | {"the wrong region", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.Region = "ie" }}, |
| 367 | {"another hold op", heldReleasePublish, nil, func(q *heldReleaseRequest) { q.HoldOp = hrRB }}, |
| 368 | {"the row held by another op", heldReleasePublish, func(f *hrFixture) { f.row.HoldOp = p1ID2 }, nil}, |
| 369 | {"no held row", heldReleasePublish, func(f *hrFixture) { f.row = nil }, nil}, |
| 370 | {"the record's move is another", heldReleasePublish, func(f *hrFixture) { f.rec.Move.ID = p1ID2 }, nil}, |
| 371 | {"the record is not at configured", heldReleasePublish, func(f *hrFixture) { f.rec.Move.LandingPhase = provision.LandLocal }, nil}, |
| 372 | {"this box is not the executor", heldReleasePublish, func(f *hrFixture) { f.claim.Op.Exec = "ie-box" }, nil}, |
| 373 | {"this box is declared lost", heldReleasePublish, func(f *hrFixture) { f.claim.Lost = []string{"uk-box"} }, nil}, |
| 374 | {"a moving claim publishes nothing", heldReleasePublish, func(f *hrFixture) { |
| 375 | f.claim.Op.Kind, f.claim.Op.Phase = hostclaim.KindMoving, "" |
| 376 | }, func(q *heldReleaseRequest) { q.Kind, q.Phase = hostclaim.KindMoving, "" }}, |
| 377 | {"a landing at local publishes nothing", heldReleasePublish, func(f *hrFixture) { |
| 378 | f.claim.Op.Phase, f.rec.Move.LandingPhase = provision.LandLocal, provision.LandLocal |
| 379 | }, func(q *heldReleaseRequest) { q.Phase = provision.LandLocal }}, |
| 380 | {"no claim", heldReleasePublish, func(f *hrFixture) { f.claim.Host = "other.example.com" }, nil}, |
| 381 | {"stage: the wrong box", heldReleaseStage, nil, func(q *heldReleaseRequest) { q.Box = "ie-box" }}, |
| 382 | {"stage: a forged app", heldReleaseStage, nil, func(q *heldReleaseRequest) { q.App = "other" }}, |
| 383 | {"stage: a forged account", heldReleaseStage, nil, func(q *heldReleaseRequest) { q.Account = "zed" }}, |
| 384 | {"stage: a stale session", heldReleaseStage, nil, func(q *heldReleaseRequest) { q.Session = 9 }}, |
| 385 | {"stage: a stale incarnation", heldReleaseStage, nil, func(q *heldReleaseRequest) { q.Incarnation = otherInc }}, |
| 386 | {"stage: changed pins", heldReleaseStage, func(f *hrFixture) { f.rec.Move.Inputs.Release = "v8" }, nil}, |
| 387 | {"stage: another target", heldReleaseStage, func(f *hrFixture) { f.claim.Op.Intent["target"] = "fr-box" }, nil}, |
| 388 | {"stage: a render hash", heldReleaseStage, nil, func(q *heldReleaseRequest) { q.RenderHash = hrRendered }}, |
| 389 | } |
| 390 | for _, tc := range cases { |
| 391 | t.Run(tc.name, func(t *testing.T) { |
| 392 | f := newHRFixture(t) |
| 393 | if tc.state != nil { |
| 394 | f.set(tc.state) |
| 395 | } |
| 396 | q := f.request(tc.action) |
| 397 | if tc.q != nil { |
| 398 | tc.q(&q) |
| 399 | } |
| 400 | _, err := f.root.Serve(ctx, hrPeer, q) |
| 401 | if err == nil { |
| 402 | t.Fatal("root accepted the request") |
| 403 | } |
| 404 | if errors.Is(err, errHeldReleaseNotYet) { |
| 405 | t.Fatalf("refused as not yet: %v", err) |
| 406 | } |
| 407 | f.untouched(t) |
| 408 | }) |
| 409 | } |
| 410 | } |
| 411 | |
| 412 | // TestHeldReleaseRootStages: a stage request stages and verifies the pinned |
| 413 | // release into the shared version store and never touches live/<host>, and |
| 414 | // selects nothing: a moving claim grants no held write. |
| 415 | func TestHeldReleaseRootStages(t *testing.T) { |
| 416 | ctx := context.Background() |
| 417 | f := newHRFixture(t) |
| 418 | f.set(func(f *hrFixture) { |
| 419 | f.claim.Op = hostclaim.Op{Kind: hostclaim.KindMoving, ID: p1ID, Exec: "ie-box", Participants: []string{"ie-box", "uk-box"}, |
| 420 | Intent: map[string]string{"source": "ie-box", "target": "uk-box", "to": "uk"}} |
| 421 | f.rec.Move.State, f.rec.Move.LandingPhase, f.rec.Move.Inputs = provision.MoveReady, "", nil |
| 422 | f.row = nil |
| 423 | }) |
| 424 | resp, err := f.root.Serve(ctx, hrPeer, f.request(heldReleaseStage)) |
| 425 | if err != nil || resp.Outcome != heldReleaseOutcomeStaged { |
| 426 | t.Fatalf("stage: %+v %v", resp, err) |
| 427 | } |
| 428 | if _, err := os.Stat(release.VersionPath(f.opt, "bab", "v7", "app")); err != nil { |
| 429 | t.Fatalf("v7 not staged: %v", err) |
| 430 | } |
| 431 | if _, err := os.Lstat(release.LiveLink(f.opt, p1Host)); !errors.Is(err, os.ErrNotExist) { |
| 432 | t.Fatalf("a stage published live/<host>: %v", err) |
| 433 | } |
| 434 | if n := f.sel.count(); n != 0 { |
| 435 | t.Fatalf("a stage selected %d time(s)", n) |
| 436 | } |
| 437 | // A publish request against a moving claim is refused: the policy is |
| 438 | // never widened. |
| 439 | q := f.request(heldReleasePublish) |
| 440 | if _, err := f.root.Serve(ctx, hrPeer, q); err == nil { |
| 441 | t.Fatal("root published under a moving claim") |
| 442 | } |
| 443 | if _, err := os.Lstat(release.LiveLink(f.opt, p1Host)); !errors.Is(err, os.ErrNotExist) { |
| 444 | t.Fatalf("live/<host> under a moving claim: %v", err) |
| 445 | } |
| 446 | } |
| 447 | |
| 448 | // TestHeldReleaseRootNotYetWithoutTheRenderJournal (platform-1e Q2): with |
| 449 | // no render-hash journal, or one without the rendered step, root refuses |
| 450 | // with the typed not-yet error at once. It selects nothing and holds no |
| 451 | // lock, so the agent's retry finds the held-stage lock free. |
| 452 | func TestHeldReleaseRootNotYetWithoutTheRenderJournal(t *testing.T) { |
| 453 | ctx := context.Background() |
| 454 | for _, tc := range []struct { |
| 455 | name string |
| 456 | prep func(t *testing.T, f *hrFixture) |
| 457 | }{ |
| 458 | {"no journal", func(t *testing.T, f *hrFixture) { |
| 459 | f.data = t.TempDir() |
| 460 | f.root.DataDir = f.data |
| 461 | }}, |
| 462 | {"no rendered step", func(t *testing.T, f *hrFixture) { |
| 463 | f.data = t.TempDir() |
| 464 | f.root.DataDir = f.data |
| 465 | if err := moves.UpdateJournal(f.data, p1Host, p1ID, func(j *moves.Journal) error { |
| 466 | j.Mark(landStepPreRender, `{"before":"absent","intended":["`+hrRendered+`"]}`) |
| 467 | return nil |
| 468 | }); err != nil { |
| 469 | t.Fatal(err) |
| 470 | } |
| 471 | }}, |
| 472 | } { |
| 473 | t.Run(tc.name, func(t *testing.T) { |
| 474 | f := newHRFixture(t) |
| 475 | tc.prep(t, f) |
| 476 | start := time.Now() |
| 477 | _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)) |
| 478 | if !errors.Is(err, errHeldReleaseNotYet) { |
| 479 | t.Fatalf("publish without the journal: %v, want errHeldReleaseNotYet", err) |
| 480 | } |
| 481 | if d := time.Since(start); d > 5*time.Second { |
| 482 | t.Fatalf("root waited %v for the journal", d) |
| 483 | } |
| 484 | f.untouched(t) |
| 485 | if err := withHeldStageLock(f.data, p1Host, func() error { return nil }); err != nil { |
| 486 | t.Fatalf("the held-stage lock after a not-yet: %v", err) |
| 487 | } |
| 488 | }) |
| 489 | } |
| 490 | } |
| 491 | |
| 492 | // TestHeldReleaseRootReplayAndCrashRecovery: a replay after success is a |
| 493 | // no-op that returns the same result (the receipt, keyed by session and |
| 494 | // render hash); a crash between the publication and its receipt is |
| 495 | // recovered by the replay, which publishes the same release again. |
| 496 | func TestHeldReleaseRootReplayAndCrashRecovery(t *testing.T) { |
| 497 | ctx := context.Background() |
| 498 | |
| 499 | t.Run("a replay after success", func(t *testing.T) { |
| 500 | f := newHRFixture(t) |
| 501 | q := f.request(heldReleasePublish) |
| 502 | first, err := f.root.Serve(ctx, hrPeer, q) |
| 503 | if err != nil { |
| 504 | t.Fatal(err) |
| 505 | } |
| 506 | reads := f.claimReads |
| 507 | q.RequestID = "req-2" |
| 508 | again, err := f.root.Serve(ctx, hrPeer, q) |
| 509 | if err != nil { |
| 510 | t.Fatalf("replay: %v", err) |
| 511 | } |
| 512 | if again.Outcome != first.Outcome || !again.Replayed { |
| 513 | t.Fatalf("replay = %+v, first = %+v", again, first) |
| 514 | } |
| 515 | if f.sel.count() != 2 || f.claimReads <= reads { |
| 516 | t.Fatalf("a replay selected %d time(s) and read the claim %d more time(s)", f.sel.count(), f.claimReads-reads) |
| 517 | } |
| 518 | }) |
| 519 | |
| 520 | t.Run("another session is not a replay", func(t *testing.T) { |
| 521 | f := newHRFixture(t) |
| 522 | if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); err != nil { |
| 523 | t.Fatal(err) |
| 524 | } |
| 525 | f.set(func(f *hrFixture) { f.claim.Holder.Session = 4 }) |
| 526 | resp, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)) |
| 527 | if err != nil || resp.Replayed { |
| 528 | t.Fatalf("a new session's request: %+v %v", resp, err) |
| 529 | } |
| 530 | if f.sel.count() != 2 { |
| 531 | t.Fatalf("selections = %d, want 2", f.sel.count()) |
| 532 | } |
| 533 | }) |
| 534 | |
| 535 | t.Run("a configured receipt never answers the probe", func(t *testing.T) { |
| 536 | // Fix round 2, P3-1: the receipt binds the action, the kind and the |
| 537 | // phase, so a landing's configured publication does not answer its |
| 538 | // probing request within the same holder tuple: the probe's request |
| 539 | // runs the whole path. |
| 540 | f := newHRFixture(t) |
| 541 | if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); err != nil { |
| 542 | t.Fatal(err) |
| 543 | } |
| 544 | f.set(func(f *hrFixture) { |
| 545 | f.claim.Op.Phase = provision.LandProbing |
| 546 | f.rec.Move.LandingPhase = provision.LandProbing |
| 547 | }) |
| 548 | reads := f.claimReads |
| 549 | resp, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)) |
| 550 | if err != nil || resp.Replayed { |
| 551 | t.Fatalf("the probe's request: %+v %v, want the whole path", resp, err) |
| 552 | } |
| 553 | if f.sel.count() != 2 || f.claimReads == reads { |
| 554 | t.Fatalf("selections %d, claim reads +%d: the probe was answered from configured's receipt", f.sel.count(), f.claimReads-reads) |
| 555 | } |
| 556 | }) |
| 557 | |
| 558 | t.Run("a stale request is not answered from an old receipt", func(t *testing.T) { |
| 559 | f := newHRFixture(t) |
| 560 | if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); err != nil { |
| 561 | t.Fatal(err) |
| 562 | } |
| 563 | q := f.request(heldReleasePublish) |
| 564 | q.Version = "v8" |
| 565 | if _, err := f.root.Serve(ctx, hrPeer, q); err == nil { |
| 566 | t.Fatal("another version was answered from the receipt") |
| 567 | } |
| 568 | }) |
| 569 | |
| 570 | t.Run("a receipt that does not persist is no success", func(t *testing.T) { |
| 571 | // Fix round 1, item 1: root answers published only once the receipt |
| 572 | // that answers a replay has persisted. Without it the answer is an |
| 573 | // error; the next identical request runs the whole path again (the |
| 574 | // same render, the same selection checks), publishing the same |
| 575 | // release, and still errs until a receipt persists. Then a replay |
| 576 | // is answered from the receipt with no selection. |
| 577 | f := newHRFixture(t) |
| 578 | blocker := filepath.Join(t.TempDir(), "not-a-dir") |
| 579 | if err := os.WriteFile(blocker, nil, 0o600); err != nil { |
| 580 | t.Fatal(err) |
| 581 | } |
| 582 | good := f.root.ReceiptDir |
| 583 | f.root.ReceiptDir = filepath.Join(blocker, "held-release") |
| 584 | q := f.request(heldReleasePublish) |
| 585 | for i := 1; i <= 2; i++ { |
| 586 | resp, err := f.root.Serve(ctx, hrPeer, q) |
| 587 | if err == nil || resp.Outcome != "" { |
| 588 | t.Fatalf("attempt %d with no receipt: %+v %v, want an error", i, resp, err) |
| 589 | } |
| 590 | if !errors.Is(err, errHeldReleaseReceipt) { |
| 591 | t.Fatalf("attempt %d: %v, want errHeldReleaseReceipt", i, err) |
| 592 | } |
| 593 | if f.sel.count() != i || len(f.sel.ids) != i || f.sel.ids[i-1] != (hostwrite.Identity{Release: "v7"}) { |
| 594 | t.Fatalf("attempt %d: selections %d under %v, want each attempt through the whole path", i, f.sel.count(), f.sel.ids) |
| 595 | } |
| 596 | } |
| 597 | // A retry is held to the same render: a changed journal refuses it |
| 598 | // before any selection. |
| 599 | f.journal(t, p1ID, hrOther) |
| 600 | if _, err := f.root.Serve(ctx, hrPeer, q); err == nil || f.sel.count() != 2 { |
| 601 | t.Fatalf("a retry under another render: %v, selections %d", err, f.sel.count()) |
| 602 | } |
| 603 | f.journal(t, p1ID, hrRendered) |
| 604 | f.root.ReceiptDir = good |
| 605 | resp, err := f.root.Serve(ctx, hrPeer, q) |
| 606 | if err != nil || resp.Outcome != heldReleaseOutcomePublished || resp.Replayed { |
| 607 | t.Fatalf("once the receipt persists: %+v %v", resp, err) |
| 608 | } |
| 609 | again, err := f.root.Serve(ctx, hrPeer, q) |
| 610 | if err != nil || !again.Replayed || f.sel.count() != 4 { |
| 611 | t.Fatalf("the replay: %+v %v, selections %d", again, err, f.sel.count()) |
| 612 | } |
| 613 | if v, _ := release.LiveVersion(f.opt, p1Host); v != "v7" { |
| 614 | t.Fatalf("live = %q", v) |
| 615 | } |
| 616 | }) |
| 617 | |
| 618 | t.Run("a crash between the publication and the receipt", func(t *testing.T) { |
| 619 | f := newHRFixture(t) |
| 620 | crashed := false |
| 621 | f.root.afterPublish = func(string) error { |
| 622 | if !crashed { |
| 623 | crashed = true |
| 624 | return errCrash |
| 625 | } |
| 626 | return nil |
| 627 | } |
| 628 | if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); !errors.Is(err, errCrash) { |
| 629 | t.Fatalf("first publish: %v, want the crash", err) |
| 630 | } |
| 631 | if v, _ := release.LiveVersion(f.opt, p1Host); v != "v7" { |
| 632 | t.Fatalf("live = %q after the crash", v) |
| 633 | } |
| 634 | if _, err := os.Stat(filepath.Join(f.receipts, p1Host+".json")); !errors.Is(err, os.ErrNotExist) { |
| 635 | t.Fatalf("a receipt before the crash point: %v", err) |
| 636 | } |
| 637 | resp, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)) |
| 638 | if err != nil || resp.Outcome != heldReleaseOutcomePublished || resp.Replayed { |
| 639 | t.Fatalf("recovery: %+v %v", resp, err) |
| 640 | } |
| 641 | if v, _ := release.LiveVersion(f.opt, p1Host); v != "v7" || f.sel.count() != 2 { |
| 642 | t.Fatalf("after recovery live = %q, selections %d", v, f.sel.count()) |
| 643 | } |
| 644 | if _, err := os.Stat(filepath.Join(f.receipts, p1Host+".json")); err != nil { |
| 645 | t.Fatalf("no receipt after recovery: %v", err) |
| 646 | } |
| 647 | }) |
| 648 | } |
| 649 | |
| 650 | // TestHeldReleaseRequestShape: root refuses a request whose shape it cannot |
| 651 | // trust before any read: the schema, the action, the names, the ids, the |
| 652 | // incarnation, the stamp and the render hash. |
| 653 | func TestHeldReleaseRequestShape(t *testing.T) { |
| 654 | f := newHRFixture(t) |
| 655 | for name, mutate := range map[string]func(q *heldReleaseRequest){ |
| 656 | "schema": func(q *heldReleaseRequest) { q.Schema = "carlos.moves.held-stage/1" }, |
| 657 | "action": func(q *heldReleaseRequest) { q.Action = "install" }, |
| 658 | "no request id": func(q *heldReleaseRequest) { q.RequestID = "" }, |
| 659 | "host": func(q *heldReleaseRequest) { q.Host = "../etc" }, |
| 660 | "account": func(q *heldReleaseRequest) { q.Account = "Bab/x" }, |
| 661 | "app": func(q *heldReleaseRequest) { q.App = "a/../b" }, |
| 662 | "version": func(q *heldReleaseRequest) { q.Version = "../v7" }, |
| 663 | "move id": func(q *heldReleaseRequest) { q.MoveID = "mv_x" }, |
| 664 | "hold op": func(q *heldReleaseRequest) { q.HoldOp = "../" + p1ID }, |
| 665 | "box": func(q *heldReleaseRequest) { q.Box = "" }, |
| 666 | "region": func(q *heldReleaseRequest) { q.Region = "UK!" }, |
| 667 | "incarnation": func(q *heldReleaseRequest) { q.Incarnation = "x" }, |
| 668 | "stamp": func(q *heldReleaseRequest) { q.Stamp = 0 }, |
| 669 | "kind": func(q *heldReleaseRequest) { q.Kind = hostclaim.KindServing }, |
| 670 | "render hash": func(q *heldReleaseRequest) { q.RenderHash = "sha256:x" }, |
| 671 | "no render hash": func(q *heldReleaseRequest) { q.RenderHash = "" }, |
| 672 | "oversized phase": func(q *heldReleaseRequest) { q.Phase = strings.Repeat("p", 200) }, |
| 673 | "oversized gen id": func(q *heldReleaseRequest) { q.Generation = strings.Repeat("g", 300) }, |
| 674 | } { |
| 675 | t.Run(name, func(t *testing.T) { |
| 676 | q := f.request(heldReleasePublish) |
| 677 | mutate(&q) |
| 678 | reads := f.claimReads |
| 679 | if _, err := f.root.Serve(context.Background(), hrPeer, q); !errors.Is(err, errHeldReleaseShape) { |
| 680 | t.Fatalf("serve: %v, want errHeldReleaseShape", err) |
| 681 | } |
| 682 | if f.claimReads != reads { |
| 683 | t.Fatal("a malformed request reached the claim") |
| 684 | } |
| 685 | }) |
| 686 | } |
| 687 | t.Run("root refuses a peer with no identity", func(t *testing.T) { |
| 688 | if _, err := f.root.Serve(context.Background(), StopPeer{}, f.request(heldReleasePublish)); !errors.Is(err, ErrClaimStopIdentity) { |
| 689 | t.Fatalf("serve: %v", err) |
| 690 | } |
| 691 | f.untouched(t) |
| 692 | }) |
| 693 | t.Run("the body is bounded and strict", func(t *testing.T) { |
| 694 | raw, _ := json.Marshal(f.request(heldReleasePublish)) |
| 695 | if _, err := decodeHeldReleaseRequest(strings.NewReader(strings.TrimSuffix(string(raw), "}") + `,"extra":1}`)); !errors.Is(err, errHeldReleaseShape) { |
| 696 | t.Fatalf("unknown field: %v", err) |
| 697 | } |
| 698 | if _, err := decodeHeldReleaseRequest(strings.NewReader(string(raw) + `{}`)); !errors.Is(err, errHeldReleaseShape) { |
| 699 | t.Fatalf("trailing data: %v", err) |
| 700 | } |
| 701 | if _, err := decodeHeldReleaseRequest(strings.NewReader(strings.Repeat(" ", heldReleaseRequestMax+1) + string(raw))); !errors.Is(err, errHeldReleaseShape) { |
| 702 | t.Fatalf("oversized: %v", err) |
| 703 | } |
| 704 | if q, err := decodeHeldReleaseRequest(strings.NewReader(string(raw))); err != nil || q.Host != p1Host { |
| 705 | t.Fatalf("a good body: %+v %v", q, err) |
| 706 | } |
| 707 | }) |
| 708 | } |
| 709 | |
| 710 | // TestHeldReleaseRootSelectionMustBeTheRequests: the claim moving on between |
| 711 | // root's read and its own selection (a new session admitted) is refused |
| 712 | // before anything is published: the selection's tuple must be the |
| 713 | // request's. A selection the claim refuses publishes nothing either. |
| 714 | func TestHeldReleaseRootSelectionMustBeTheRequests(t *testing.T) { |
| 715 | ctx := context.Background() |
| 716 | t.Run("the claim moved on", func(t *testing.T) { |
| 717 | f := newHRFixture(t) |
| 718 | works := f.root.Works |
| 719 | f.root.Works = func(ctx context.Context, actor hostbox.Actor, host, action string) (hostwrite.Work, error) { |
| 720 | f.set(func(f *hrFixture) { f.claim.Holder.Session = 4 }) |
| 721 | return works(ctx, actor, host, action) |
| 722 | } |
| 723 | if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); !errors.Is(err, hostbox.ErrChanged) { |
| 724 | t.Fatalf("publish across a session change: %v, want hostbox.ErrChanged", err) |
| 725 | } |
| 726 | if _, err := os.Lstat(release.LiveLink(f.opt, p1Host)); !errors.Is(err, os.ErrNotExist) { |
| 727 | t.Fatalf("published across a session change: %v", err) |
| 728 | } |
| 729 | if len(f.sel.ids) != 0 { |
| 730 | t.Fatalf("wrote under %v", f.sel.ids) |
| 731 | } |
| 732 | }) |
| 733 | t.Run("the selection is refused", func(t *testing.T) { |
| 734 | f := newHRFixture(t) |
| 735 | f.sel.refuse = hostbox.ErrChanged |
| 736 | if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); !errors.Is(err, hostbox.ErrChanged) { |
| 737 | t.Fatalf("publish with the selection refused: %v", err) |
| 738 | } |
| 739 | if _, err := os.Lstat(release.LiveLink(f.opt, p1Host)); !errors.Is(err, os.ErrNotExist) { |
| 740 | t.Fatalf("published with the selection refused: %v", err) |
| 741 | } |
| 742 | }) |
| 743 | } |
| 744 | |
| 745 | // TestHeldReleaseRouteIsMovesOnly: with CARLOS_MOVES_PHASE1 off the claim |
| 746 | // root's handler is returned unchanged, so the route refuses as on main. |
| 747 | func TestHeldReleaseRouteIsMovesOnly(t *testing.T) { |
| 748 | h := ClaimRootWithLaunch{} |
| 749 | got, err := WithProductionMovesHeldRelease(context.Background(), h, nil, func(string) string { return "" }) |
| 750 | if err != nil { |
| 751 | t.Fatal(err) |
| 752 | } |
| 753 | if r, ok := got.(ClaimRootWithLaunch); !ok || r.HeldRelease != nil { |
| 754 | t.Fatalf("moves off wired %#v", got) |
| 755 | } |
| 756 | } |
| 757 | |
| 758 | // TestHeldReleaseRootRefusesUnjournalledRecoveryProbes (fix round 2, P2-1, |
| 759 | // platform-1e's authority review): a revert's probe and a clean return's |
| 760 | // probe journal no render under any rule root checks, so their absent |
| 761 | // journal cannot be what refuses them: the journal is agent-written. Root |
| 762 | // refuses both rows outright, even with a planted journal and a matching |
| 763 | // request, and publishes nothing. A dirty return's probe, whose configured |
| 764 | // step journalled its render, still publishes. |
| 765 | func TestHeldReleaseRootRefusesUnjournalledRecoveryProbes(t *testing.T) { |
| 766 | ctx := context.Background() |
| 767 | recovery := func(f *hrFixture) { |
| 768 | f.claim.Op.Exec, f.claim.Op.Participants = "uk-box", []string{"uk-box", "ie-box"} |
| 769 | f.claim.Op.Intent = map[string]string{"source": "uk-box", "target": "ie-box", "to": "ie"} |
| 770 | f.claim.Op.Phase = hostclaim.PhaseProbing |
| 771 | f.rec.Move.From, f.rec.Move.To, f.rec.Move.LandingPhase = "uk", "ie", "" |
| 772 | } |
| 773 | for _, tc := range []struct { |
| 774 | name string |
| 775 | set func(f *hrFixture) |
| 776 | hold string |
| 777 | }{ |
| 778 | {"a revert's probe", func(f *hrFixture) { |
| 779 | recovery(f) |
| 780 | f.claim.Op.Kind = hostclaim.KindReverting |
| 781 | f.claim.Op.Journal = map[string]string{hostclaim.JournalRevPhase: hostclaim.PhaseProbing} |
| 782 | f.rec.Move.State = provision.MoveReverting |
| 783 | f.rec.Move.RecoveryInputs = &provision.MoveInputs{Release: "v7"} |
| 784 | }, p1ID}, |
| 785 | {"a clean return's probe", func(f *hrFixture) { |
| 786 | recovery(f) |
| 787 | f.claim.Op.Kind = hostclaim.KindRestoring |
| 788 | f.claim.Op.Intent["rollback"] = hrRB |
| 789 | f.rec.Move.State = provision.MoveRollingBack |
| 790 | f.rec.Move.Rollback = &provision.MoveRollback{ID: hrRB, SourcePhase: provision.RbSourceProbing, RecoveryInputs: &provision.MoveInputs{Release: "v7"}} |
| 791 | f.row.HoldOp = hrRB |
| 792 | }, hrRB}, |
| 793 | } { |
| 794 | t.Run(tc.name+" with a planted journal", func(t *testing.T) { |
| 795 | f := newHRFixture(t) |
| 796 | f.set(tc.set) |
| 797 | // A compromised agent plants the render it then sends. |
| 798 | f.journal(t, tc.hold, hrRendered) |
| 799 | q := f.request(heldReleasePublish) |
| 800 | if q.HoldOp != tc.hold { |
| 801 | t.Fatalf("request hold op %q, want %q", q.HoldOp, tc.hold) |
| 802 | } |
| 803 | _, err := f.root.Serve(ctx, hrPeer, q) |
| 804 | if !errors.Is(err, release.ErrStaleHeldStage) { |
| 805 | t.Fatalf("publish: %v, want a refusal", err) |
| 806 | } |
| 807 | f.untouched(t) |
| 808 | }) |
| 809 | } |
| 810 | } |
| 811 | |
| 812 | // TestHeldReleaseRootBoundsConcurrency (fix round 2, P3-4): root serves |
| 813 | // one request per host at a time, and at most heldReleaseConcurrency in |
| 814 | // all, so the agent cannot pile up downloads inside instance-control. A |
| 815 | // request over either bound is refused at once as busy (503, which the |
| 816 | // agent's client reads as unavailable and retries), before any read; a |
| 817 | // slot is free again once its request returns. |
| 818 | func TestHeldReleaseRootBoundsConcurrency(t *testing.T) { |
| 819 | ctx := context.Background() |
| 820 | hold := func(f *hrFixture) (entered chan string, release func()) { |
| 821 | entered = make(chan string, 16) |
| 822 | gate := make(chan struct{}) |
| 823 | claims := f.root.Claims |
| 824 | f.root.Claims = func(ctx context.Context, host string) (hostclaim.Observed, error) { |
| 825 | entered <- host |
| 826 | <-gate |
| 827 | return claims(ctx, host) |
| 828 | } |
| 829 | var once sync.Once |
| 830 | release = func() { once.Do(func() { close(gate) }) } |
| 831 | t.Cleanup(release) |
| 832 | return entered, release |
| 833 | } |
| 834 | busy := func(t *testing.T, err error) { |
| 835 | t.Helper() |
| 836 | if !errors.Is(err, errHeldReleaseBusy) { |
| 837 | t.Fatalf("over the bound: %v, want errHeldReleaseBusy", err) |
| 838 | } |
| 839 | if got := heldReleaseStatus(err); got != 503 { |
| 840 | t.Fatalf("busy answers %d, want 503", got) |
| 841 | } |
| 842 | } |
| 843 | |
| 844 | t.Run("one per host", func(t *testing.T) { |
| 845 | f := newHRFixture(t) |
| 846 | entered, release := hold(f) |
| 847 | done := make(chan error, 1) |
| 848 | go func() { |
| 849 | _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)) |
| 850 | done <- err |
| 851 | }() |
| 852 | <-entered |
| 853 | reads := f.claimReads |
| 854 | _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleaseStage)) |
| 855 | busy(t, err) |
| 856 | if f.claimReads != reads || len(entered) != 0 { |
| 857 | t.Fatal("a busy request read the claim") |
| 858 | } |
| 859 | release() |
| 860 | if err := <-done; err != nil { |
| 861 | t.Fatalf("the first request: %v", err) |
| 862 | } |
| 863 | if _, err := f.root.Serve(ctx, hrPeer, f.request(heldReleasePublish)); err != nil { |
| 864 | t.Fatalf("after the first returned: %v", err) |
| 865 | } |
| 866 | }) |
| 867 | |
| 868 | t.Run("a bound across hosts", func(t *testing.T) { |
| 869 | f := newHRFixture(t) |
| 870 | entered, release := hold(f) |
| 871 | other := func(i int) heldReleaseRequest { |
| 872 | q := f.request(heldReleaseStage) |
| 873 | q.Host = fmt.Sprintf("h%d.go.example.com", i) |
| 874 | return q |
| 875 | } |
| 876 | done := make(chan error, heldReleaseConcurrency) |
| 877 | for i := 0; i < heldReleaseConcurrency; i++ { |
| 878 | go func(i int) { |
| 879 | _, err := f.root.Serve(ctx, hrPeer, other(i)) |
| 880 | done <- err |
| 881 | }(i) |
| 882 | <-entered |
| 883 | } |
| 884 | _, err := f.root.Serve(ctx, hrPeer, other(99)) |
| 885 | busy(t, err) |
| 886 | release() |
| 887 | for i := 0; i < heldReleaseConcurrency; i++ { |
| 888 | if err := <-done; errors.Is(err, errHeldReleaseBusy) { |
| 889 | t.Fatalf("a held request: %v", err) |
| 890 | } |
| 891 | } |
| 892 | if _, err := f.root.Serve(ctx, hrPeer, other(99)); errors.Is(err, errHeldReleaseBusy) { |
| 893 | t.Fatalf("after the slots freed: %v", err) |
| 894 | } |
| 895 | }) |
| 896 | } |
| 897 | |
amadan