amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1//go:build linux
2
3package hostexec
4
5import (
6 "context"
7 "encoding/json"
8 "errors"
9 "os"
10 "os/exec"
11 "path/filepath"
12 "testing"
13 "time"
14
15 "github.com/carlosframework/platform/internal/release"
16 "golang.org/x/sys/unix"
17)
18
19type brokerDisk struct {
20 dir string
21 writes int
22 failAt int
23 afterRename func(string, []byte)
24}
25
26func (d *brokerDisk) Read(n string) ([]byte, error) { return os.ReadFile(filepath.Join(d.dir, n)) }
27func (d *brokerDisk) Write(n string, b []byte) error {
28 d.writes++
29 f, err := os.CreateTemp(d.dir, "pending-")
30 if err != nil {
31 return err
32 }
33 name := f.Name()
34 defer os.Remove(name)
35 if _, err = f.Write(b); err == nil {
36 err = f.Sync()
37 }
38 err = errors.Join(err, f.Close())
39 if err != nil {
40 return err
41 }
42 if err = os.Rename(name, filepath.Join(d.dir, n)); err != nil {
43 return err
44 }
45 if d.afterRename != nil {
46 d.afterRename(n, b)
47 }
48 if d.writes == d.failAt {
49 return os.ErrInvalid
50 }
51 parent, err := os.Open(d.dir)
52 if err != nil {
53 return err
54 }
55 return errors.Join(parent.Sync(), parent.Close())
56}
57func (d *brokerDisk) Remove(n string) error { return os.Remove(filepath.Join(d.dir, n)) }
58func (d *brokerDisk) Names() ([]string, error) {
59 es, e := os.ReadDir(d.dir)
60 var out []string
61 for _, v := range es {
62 out = append(out, v.Name())
63 }
64 return out, e
65}
66func (d *brokerDisk) Lock(ctx context.Context, name string) (func(), error) {
67 f, err := os.OpenFile(filepath.Join(d.dir, name), os.O_CREATE|os.O_RDWR, 0600)
68 if err != nil {
69 return nil, err
70 }
71 for {
72 if err := ctx.Err(); err != nil {
73 f.Close()
74 return nil, err
75 }
76 err = unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB)
77 if err == nil {
78 return func() { _ = unix.Flock(int(f.Fd()), unix.LOCK_UN); _ = f.Close() }, nil
79 }
80 if !errors.Is(err, unix.EWOULDBLOCK) && !errors.Is(err, unix.EAGAIN) {
81 f.Close()
82 return nil, err
83 }
84 select {
85 case <-ctx.Done():
86 f.Close()
87 return nil, ctx.Err()
88 case <-time.After(time.Millisecond):
89 }
90 }
91}
92func (d *brokerDisk) Close() error { return nil }
93
94func brokerFixture(t *testing.T) (*RootBroker, *brokerDisk, *delayedStore, *testClock, ExecutionRequest, ControllerProcess) {
95 t.Helper()
96 return brokerFixtureAt(t, t.TempDir(), t.TempDir())
97}
98func brokerFixtureAt(t *testing.T, local, remotePath string) (*RootBroker, *brokerDisk, *delayedStore, *testClock, ExecutionRequest, ControllerProcess) {
99 t.Helper()
100 d := &brokerDisk{dir: local}
101 clock := &testClock{wall: time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC), boot: int64(100 * time.Second)}
102 remote := &delayedStore{Store: release.NewDirStore(remotePath)}
103 q := ExecutionRequest{Schema: 1, ID: "begin-1", Operation: "executionBegin", Host: "shop.example.com", Incarnation: "inc", Stamp: 3, Session: 2, Op: "move-1"}
104 p := ControllerProcess{KernelBootID: "boot", PID: 42, Start: "start-42"}
105 b, err := NewRootBroker(RootBrokerConfig{State: d, Remote: remote, Clock: clock, BootID: "boot", Identity: "broker-1",
106 Authority: func(_ context.Context, got ExecutionRequest, peer ControllerProcess) (ExecutionAuthority, error) {
107 if got.Host != q.Host || got.Incarnation != q.Incarnation || got.Stamp != q.Stamp || got.Session != q.Session || got.Op != q.Op || peer != p {
108 return ExecutionAuthority{}, ErrInactive
109 }
110 return ExecutionAuthority{Box: "box", Enrollment: "en", Account: "bab", App: "jam", Installation: "installation-1", Migration: 1, ClaimRevision: "claim-1", RecordRevision: "record-1", DirectoryRevision: "directory-1", Kind: "landing", Phase: "checking"}, nil
111 }, Alive: func(context.Context, ControllerProcess) error { return nil }, Cleanup: func(context.Context, ExecutionRecord) error { return nil }, Settle: func(context.Context, ExecutionRecord) error { return nil },
112 Exclude: func(_ context.Context, fn func() error) error { return fn() },
113 })
114 if err != nil {
115 t.Fatal(err)
116 }
117 t.Cleanup(func() { _ = b.Close(context.Background()) })
118 return b, d, remote, clock, q, p
119}
120
121func TestPlan14ExecutionBrokerPeerAndShape(t *testing.T) {
122 for _, mutation := range []string{"schema", "host", "incarnation", "stamp", "session", "operation", "peer-start", "peer-pid", "peer-boot"} {
123 t.Run(mutation, func(t *testing.T) {
124 b, d, remote, _, q, p := brokerFixture(t)
125 switch mutation {
126 case "schema":
127 q.Schema++
128 case "host":
129 q.Host = "other.example.com"
130 case "incarnation":
131 q.Incarnation = "other"
132 case "stamp":
133 q.Stamp++
134 case "session":
135 q.Session++
136 case "operation":
137 q.Op = "other"
138 case "peer-start":
139 p.Start = "reused"
140 case "peer-pid":
141 p.PID++
142 case "peer-boot":
143 p.KernelBootID = "reboot"
144 }
145 if _, err := b.Execute(t.Context(), p, q); err == nil {
146 t.Fatal("altered identity acquired participation")
147 }
148 if remote.calls != 0 || d.writes != 0 {
149 t.Fatalf("refusal had effects: remote=%d writes=%d", remote.calls, d.writes)
150 }
151 })
152 }
153 b, d, remote, _, q, p := brokerFixture(t)
154 first, err := b.Execute(t.Context(), p, q)
155 if err != nil {
156 t.Fatal(err)
157 }
158 writes := d.writes
159 again, err := b.Execute(t.Context(), p, q)
160 if err != nil || first != again {
161 t.Fatalf("replay: %+v %+v %v", first, again, err)
162 }
163 if remote.calls != 1 || d.writes != writes {
164 t.Fatal("replay renewed or rewrote participation")
165 }
166 q.ID = "another-controller"
167 if _, err = b.Execute(t.Context(), p, q); err == nil {
168 t.Fatal("another controller borrowed token")
169 }
170}
171
172func TestPlan14ExecutionDeadlineFromRequestStart(t *testing.T) {
173 for _, delay := range []time.Duration{30 * time.Second, 85 * time.Second, 100 * time.Second} {
174 t.Run(delay.String(), func(t *testing.T) {
175 b, _, remote, clock, q, p := brokerFixture(t)
176 remote.onPut = func(int) { clock.advance(delay) }
177 out, err := b.Execute(t.Context(), p, q)
178 if delay >= 85*time.Second {
179 if err == nil {
180 t.Fatal("late response granted participation")
181 }
182 return
183 }
184 if err != nil {
185 t.Fatal(err)
186 }
187 if out.SafeUntil != int64(185*time.Second) {
188 t.Fatalf("response-relative deadline: %v", out)
189 }
190 clock.advance(55 * time.Second)
191 q.Operation = "executionView"
192 q.ID = "view-1"
193 q.Token = out.Token
194 if _, err = b.Execute(t.Context(), p, q); err == nil {
195 t.Fatal("suspend beyond cutoff retained permission")
196 }
197 })
198 }
199}
200
201func TestPlan14ExecutionPendingSurvivesCommitFailure(t *testing.T) {
202 b, d, remote, _, q, p := brokerFixture(t)
203 d.failAt = 2
204 if _, err := b.Execute(t.Context(), p, q); err == nil {
205 t.Fatal("failed commit replied successfully")
206 }
207 if remote.calls != 1 {
208 t.Fatal("did not cross remote acquisition")
209 }
210 if _, err := b.Execute(t.Context(), p, q); err == nil {
211 t.Fatal("uncertain commit replay granted permission")
212 }
213 if remote.calls != 1 {
214 t.Fatal("uncertain replay acquired twice")
215 }
216 var record ExecutionRecord
217 raw, err := d.Read(q.Host + ".json")
218 if err != nil {
219 t.Fatal(err)
220 }
221 if err = json.Unmarshal(raw, &record); err != nil {
222 t.Fatal(err)
223 }
224 if record.Status == "settled" {
225 t.Fatal("uncertain acquisition lost obligation")
226 }
227}
228
229func TestPlan14ExecutionBrokerRestartCannotResume(t *testing.T) {
230 b, _, _, _, q, p := brokerFixture(t)
231 permit, err := b.Execute(t.Context(), p, q)
232 if err != nil {
233 t.Fatal(err)
234 }
235 if err := b.Close(t.Context()); err != nil {
236 t.Fatal(err)
237 }
238 cfg := b.cfg
239 cfg.Identity = "broker-2"
240 restarted, err := NewRootBroker(cfg)
241 if err != nil {
242 t.Fatal(err)
243 }
244 defer restarted.Close(t.Context())
245 q.Operation = "executionView"
246 q.Token = permit.Token
247 q.ID = "view-1"
248 if _, err = restarted.Execute(t.Context(), p, q); !errors.Is(err, ErrLost) {
249 t.Fatalf("restart restored prior permit: %v", err)
250 }
251}
252
253func TestPlan14ExecutionAuthorityLossIsSticky(t *testing.T) {
254 b, _, _, _, q, p := brokerFixture(t)
255 out, err := b.Execute(t.Context(), p, q)
256 if err != nil {
257 t.Fatal(err)
258 }
259 original := b.cfg.Authority
260 b.cfg.Authority = func(context.Context, ExecutionRequest, ControllerProcess) (ExecutionAuthority, error) {
261 return ExecutionAuthority{}, ErrInactive
262 }
263 q.Operation = "executionView"
264 q.Token = out.Token
265 q.ID = "changed"
266 if _, err = b.Execute(t.Context(), p, q); err == nil {
267 t.Fatal("changed authority remained active")
268 }
269 b.cfg.Authority = original
270 q.ID = "restored"
271 if _, err = b.Execute(t.Context(), p, q); err == nil {
272 t.Fatal("restoring authority resurrected lost lease")
273 }
274}
275
276func TestPlan14ExecutionInvalidAuthorityAndReplayAreSticky(t *testing.T) {
277 for _, name := range []string{"invalid", "replay"} {
278 t.Run(name, func(t *testing.T) {
279 replay := name == "replay"
280 b, _, _, _, q, peer := brokerFixture(t)
281 initial, err := b.Execute(t.Context(), peer, q)
282 if err != nil {
283 t.Fatal(err)
284 }
285 original := b.cfg.Authority
286 b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) {
287 a, err := original(ctx, q, p)
288 if replay {
289 a.Phase = "changed"
290 } else {
291 a.Migration = 0
292 }
293 return a, err
294 }
295 if !replay {
296 q.Operation, q.ID, q.Token = "executionView", "invalid-authority", initial.Token
297 }
298 if _, err := b.Execute(t.Context(), peer, q); err == nil {
299 t.Fatal("invalid authority accepted")
300 }
301 b.cfg.Authority = original
302 q.Operation, q.ID, q.Token = "executionView", "after-repair", initial.Token
303 if _, err := b.Execute(t.Context(), peer, q); err == nil {
304 t.Fatal("authority repair resurrected denied participation", replay)
305 }
306 })
307 }
308}
309
310func TestPlan14ExecutionUncertainReleaseCanSettleAfterExpiry(t *testing.T) {
311 b, d, remote, clock, q, p := brokerFixture(t)
312 remote.onPut = func(int) { d.failAt = d.writes + 1 }
313 if _, err := b.Execute(t.Context(), p, q); err == nil {
314 t.Fatal("failed acquisition committed")
315 }
316 r, err := b.read(q.Host)
317 if err != nil {
318 t.Fatal(err)
319 }
320 // Simulate the crash window with no retained conditional-write response.
321 r.LeaseETag = ""
322 r.Status = "pending"
323 if err = b.write(r); err != nil {
324 t.Fatal(err)
325 }
326 q.Operation = "executionSettle"
327 q.Token = r.Token
328 q.ID = "settle"
329 if _, err = b.Execute(t.Context(), p, q); err == nil {
330 t.Fatal("unexpired uncertain lease settled")
331 }
332 clock.advance(TTL + time.Second)
333 if _, err = b.Execute(t.Context(), p, q); err != nil {
334 t.Fatal("expired exact obligation cannot settle", err)
335 }
336}
337
338func TestPlan14ExecutionMonitorIndependentOfRenewal(t *testing.T) {
339 b, _, remote, clock, q, p := brokerFixture(t)
340 out, err := b.Execute(t.Context(), p, q)
341 if err != nil {
342 t.Fatal(err)
343 }
344 entered, releasePut := make(chan struct{}), make(chan struct{})
345 remote.onPut = func(int) { close(entered); <-releasePut }
346 cleaned := make(chan struct{}, 1)
347 b.cfg.Cleanup = func(context.Context, ExecutionRecord) error {
348 select {
349 case cleaned <- struct{}{}:
350 default:
351 }
352 return nil
353 }
354 done := make(chan error, 1)
355 q.Operation = "executionRenew"
356 q.ID = "renew"
357 q.Token = out.Token
358 go func() { _, err := b.Execute(context.Background(), p, q); done <- err }()
359 <-entered
360 clock.advance(TTL)
361 monitored := make(chan error, 1)
362 go func() { monitored <- b.Monitor(context.Background()) }()
363 select {
364 case <-cleaned:
365 case <-time.After(time.Second):
366 close(releasePut)
367 <-done
368 t.Fatal("network renewal blocked deadline cleanup")
369 }
370 close(releasePut)
371 if err = <-done; err == nil {
372 t.Fatal("late renewal revived expired lease")
373 }
374 <-monitored
375}
376
377func TestPlan14ExecutionBindsRegisteredInstallation(t *testing.T) {
378 b, _, _, _, q, p := brokerFixture(t)
379 base := b.cfg.Authority
380 installed := false
381 b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) {
382 a, e := base(ctx, q, p)
383 if !installed {
384 a.Installation = ""
385 a.ParticipationOnly = true
386 }
387 return a, e
388 }
389 out, err := b.Execute(t.Context(), p, q)
390 if err != nil {
391 t.Fatal(err)
392 }
393 installed = true
394 q.Operation = "executionView"
395 q.ID = "registered"
396 q.Token = out.Token
397 if _, err = b.Execute(t.Context(), p, q); err != nil {
398 t.Fatal("root registration lost participation", err)
399 }
400 r, err := b.read(q.Host)
401 if err != nil {
402 t.Fatal(err)
403 }
404 if r.Authority.Installation == "" || r.Authority.ParticipationOnly {
405 t.Fatal("lease was not bound to root installation")
406 }
407 installed = false
408 q.ID = "removed"
409 if _, err = b.Execute(t.Context(), p, q); err == nil {
410 t.Fatal("removed root installation retained participation")
411 }
412}
413
414func TestPlan14ExecutionRootReadIgnoresMirrors(t *testing.T) {
415 b, d, _, _, q, p := brokerFixture(t)
416 original, oldPeer := b.cfg.Authority, p
417 var err error
418 p, err = currentControllerProcess()
419 if err != nil {
420 t.Fatal(err)
421 }
422 b.cfg.BootID = p.KernelBootID
423 b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, _ ControllerProcess) (ExecutionAuthority, error) {
424 return original(ctx, q, oldPeer)
425 }
426 out, err := b.Execute(t.Context(), p, q)
427 if err != nil {
428 t.Fatal(err)
429 }
430 census, _ := json.Marshal(map[string]any{"Schema": 1, "Box": "box", "Enrollment": "en", "Migration": 1, "Revision": 1, "LaunchRevision": 1, "Complete": true})
431 if err = d.Write("census.json", census); err != nil {
432 t.Fatal(err)
433 }
434 if _, err = ReadRoot(t.Context(), d, b.cfg.Clock, q, false); err != nil {
435 t.Fatal("valid root permit refused", err)
436 }
437 saved, err := b.read(q.Host)
438 if err != nil {
439 t.Fatal(err)
440 }
441 only := saved
442 only.Authority.ParticipationOnly = true
443 if err := b.write(only); err != nil {
444 t.Fatal(err)
445 }
446 if _, err := ReadRoot(t.Context(), d, b.cfg.Clock, q, false); err == nil {
447 t.Fatal("participation-only record granted privileged execution")
448 }
449 if err := b.write(saved); err != nil {
450 t.Fatal(err)
451 }
452 data := t.TempDir()
453 mirror := filepath.Join(data, ".hostexec")
454 if err = os.Mkdir(mirror, 0700); err != nil {
455 t.Fatal(err)
456 }
457 if err = os.WriteFile(filepath.Join(mirror, q.Host+".json"), []byte(`{"safe_until_boottime_ns":999999999999999999,"state":"active"}`), 0600); err != nil {
458 t.Fatal(err)
459 }
460 if err = os.Rename(mirror, filepath.Join(data, "old")); err != nil {
461 t.Fatal(err)
462 }
463 if err = os.Mkdir(mirror, 0700); err != nil {
464 t.Fatal(err)
465 }
466 if _, err = ReadRoot(t.Context(), d, b.cfg.Clock, q, false); err != nil {
467 t.Fatal("service replacement changed root permit", err)
468 }
469 // A failed durable loss still drops the root-owned live lock.
470 d.failAt = d.writes + 1
471 r, err := b.read(q.Host)
472 if err != nil {
473 t.Fatal(err)
474 }
475 if err = b.deny(t.Context(), r, ErrLost); err == nil {
476 t.Fatal("lost lease reported success")
477 }
478 r.Status = "active"
479 r.Token = out.Token
480 raw, _ := json.Marshal(r)
481 if err = os.WriteFile(filepath.Join(d.dir, q.Host+".json"), raw, 0600); err != nil {
482 t.Fatal(err)
483 }
484 if _, err = ReadRoot(t.Context(), d, b.cfg.Clock, q, false); err == nil {
485 t.Fatal("old active bytes revived revoked live lock")
486 }
487}
488
489func TestPlan14ExecutionLossPersistenceRevokesAllLiveLocks(t *testing.T) {
490 b, d, _, _, q, peer := brokerFixture(t)
491 original := b.cfg.Authority
492 b.cfg.Authority = func(ctx context.Context, request ExecutionRequest, peer ControllerProcess) (ExecutionAuthority, error) {
493 request.Host = q.Host
494 return original(ctx, request, peer)
495 }
496 if _, err := b.Execute(t.Context(), peer, q); err != nil {
497 t.Fatal(err)
498 }
499 other := q
500 other.Host = "other.example.com"
501 if _, err := b.Execute(t.Context(), peer, other); err != nil {
502 t.Fatal(err)
503 }
504 r, err := b.read(q.Host)
505 if err != nil {
506 t.Fatal(err)
507 }
508 d.failAt = d.writes + 1
509 if err := b.deny(t.Context(), r, ErrLost); err == nil {
510 t.Fatal("loss persistence failure accepted")
511 }
512 held, err := d.LiveHeld(other.Host + ".live")
513 if err != nil || held {
514 t.Fatal("independent reader still sees live permit", held, err)
515 }
516
517}
518
519func TestPlan14ExecutionVisibleRenameCannotGrantBeforeSync(t *testing.T) {
520 b, d, _, clock, q, priorPeer := brokerFixture(t)
521 peer, err := currentControllerProcess()
522 if err != nil {
523 t.Fatal(err)
524 }
525 b.cfg.BootID = peer.KernelBootID
526 original := b.cfg.Authority
527 b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, _ ControllerProcess) (ExecutionAuthority, error) {
528 return original(ctx, q, priorPeer)
529 }
530 initial, err := b.Execute(t.Context(), peer, q)
531 if err != nil {
532 t.Fatal(err)
533 }
534 census, _ := json.Marshal(map[string]any{"Schema": 1, "Box": "box", "Enrollment": "en", "Migration": 1, "Revision": 1, "LaunchRevision": 1, "Complete": true})
535 if err := d.Write("census.json", census); err != nil {
536 t.Fatal(err)
537 }
538 if _, err := ReadRoot(t.Context(), d, clock, q, false); err != nil {
539 t.Fatal("initial permit", err)
540 }
541 clock.advance(Renewal)
542 visible, finish := make(chan struct{}), make(chan struct{})
543 d.afterRename = func(_ string, raw []byte) {
544 var r ExecutionRecord
545 if json.Unmarshal(raw, &r) == nil && r.Status == "active" {
546 close(visible)
547 <-finish
548 }
549 }
550 q.Operation, q.ID, q.Token = "executionRenew", "renew-before-sync", initial.Token
551 done := make(chan error, 1)
552 go func() { _, err := b.Execute(t.Context(), peer, q); done <- err }()
553 select {
554 case <-visible:
555 case <-time.After(time.Second):
556 t.Fatal("renew did not reach visible rename")
557 }
558 _, readErr := ReadRoot(t.Context(), d, clock, q, false)
559 close(finish)
560 if err := <-done; err != nil {
561 t.Fatal(err)
562 }
563 if readErr == nil {
564 t.Fatal("visible but unsynced active bytes granted execution")
565 }
566 if _, err := ReadRoot(t.Context(), d, clock, q, false); err != nil {
567 t.Fatal("durable renewal refused", err)
568 }
569}
570
571func TestPlan14ExecutionSettledTombstoneDoesNotBlockNextOperation(t *testing.T) {
572 b, _, _, _, q, p := brokerFixture(t)
573 out, err := b.Execute(t.Context(), p, q)
574 if err != nil {
575 t.Fatal(err)
576 }
577 settle := q
578 settle.ID = "settle"
579 settle.Operation = "executionSettle"
580 settle.Token = out.Token
581 if _, err = b.Execute(t.Context(), p, settle); err != nil {
582 t.Fatal(err)
583 }
584 next := q
585 next.ID = "next-begin"
586 next.Op = "move-2"
587 original := b.cfg.Authority
588 b.cfg.Authority = func(ctx context.Context, got ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) {
589 copy := got
590 copy.Op = q.Op
591 return original(ctx, copy, p)
592 }
593 granted, err := b.Execute(t.Context(), p, next)
594 if err != nil {
595 t.Fatal("settled operation permanently blocked host", err)
596 }
597 if granted.Token == out.Token {
598 t.Fatal("new operation reused old permission")
599 }
600 if old, err := b.Execute(t.Context(), p, settle); err != nil || old.Status != "settled" {
601 t.Fatalf("settlement tombstone lost: %+v %v", old, err)
602 }
603 if _, err = b.Execute(t.Context(), p, q); err == nil {
604 t.Fatal("old begin resurrected")
605 }
606}
607
608func TestPlan14ExecutionSessionTransitionRetainsRemoteOwner(t *testing.T) {
609 b, _, remote, _, q, p := brokerFixture(t)
610 original := b.cfg.Authority
611 changed := false
612 b.cfg.Authority = func(ctx context.Context, got ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) {
613 expected := q.Session
614 if changed {
615 expected++
616 }
617 if got.Session != expected {
618 return ExecutionAuthority{}, ErrInactive
619 }
620 copy := got
621 copy.Session = q.Session
622 a, e := original(ctx, copy, p)
623 if changed {
624 a.ClaimRevision = "claim-2"
625 a.Phase = "configured"
626 }
627 return a, e
628 }
629 b.cfg.Transition = func(_ context.Context, old ExecutionRecord, next ExecutionRequest, a ExecutionAuthority) error {
630 if old.Request.Session+1 != next.Session || old.Authority.Installation != a.Installation {
631 return ErrLost
632 }
633 return nil
634 }
635 first, err := b.Execute(t.Context(), p, q)
636 if err != nil {
637 t.Fatal(err)
638 }
639 changed = true
640 next := q
641 next.ID = "new-session"
642 next.Operation = "executionRenew"
643 next.Session++
644 next.Token = first.Token
645 second, err := b.Execute(t.Context(), p, next)
646 if err != nil {
647 t.Fatal(err)
648 }
649 if second.Token == first.Token || second.Session != next.Session || remote.calls != 2 {
650 t.Fatalf("transition: %+v calls=%d", second, remote.calls)
651 }
652 again, err := b.Execute(t.Context(), p, next)
653 if err != nil || again != second || remote.calls != 2 {
654 t.Fatalf("lost transition response replay: %+v %v", again, err)
655 }
656 stale := next
657 stale.Operation = "executionView"
658 stale.ID = "stale-view"
659 if _, err = b.Execute(t.Context(), p, stale); err == nil {
660 t.Fatal("old token admitted new session")
661 }
662}
663
664type executionCrashStore struct{ release.Store }
665
666func (s executionCrashStore) Put(ctx context.Context, key string, raw []byte, etag string) (string, error) {
667 next, err := s.Store.Put(ctx, key, raw, etag)
668 if err == nil {
669 os.Exit(73)
670 }
671 return next, err
672}
673
674func TestPlan14ExecutionProcessCrashBeforeCommit(t *testing.T) {
675 if dir := os.Getenv("PLAN14_EXECUTION_CRASH_DIR"); dir != "" {
676 b, _, remote, _, q, p := brokerFixtureAt(t, filepath.Join(dir, "local"), filepath.Join(dir, "remote"))
677 b.cfg.Remote = executionCrashStore{remote.Store}
678 _, _ = b.Execute(context.Background(), p, q)
679 t.Fatal("crash boundary was not reached")
680 }
681 dir := t.TempDir()
682 for _, name := range []string{"local", "remote"} {
683 if err := os.Mkdir(filepath.Join(dir, name), 0700); err != nil {
684 t.Fatal(err)
685 }
686 }
687 child := exec.Command(os.Args[0], "-test.run=^TestPlan14ExecutionProcessCrashBeforeCommit$", "-test.count=1")
688 child.Dir = dir
689 child.Env = append(os.Environ(), "PLAN14_EXECUTION_CRASH_DIR="+dir)
690 raw, err := child.CombinedOutput()
691 var exited *exec.ExitError
692 if !errors.As(err, &exited) || exited.ExitCode() != 73 {
693 t.Fatalf("child: %v %s", err, raw)
694 }
695 b, _, remote, clock, q, p := brokerFixtureAt(t, filepath.Join(dir, "local"), filepath.Join(dir, "remote"))
696 b.cfg.Identity = "restarted-broker"
697 record, err := b.read(q.Host)
698 if err != nil {
699 t.Fatal(err)
700 }
701 if record.Status != "pending" || record.LeaseETag != "" {
702 t.Fatalf("crash did not retain uncertain pending: %+v", record)
703 }
704 key, _ := Key(q.Host, record.Authority.Box)
705 if _, _, err = remote.Get(t.Context(), key); err != nil {
706 t.Fatal("remote acquisition missing", err)
707 }
708 if _, err = b.Execute(t.Context(), p, q); err == nil {
709 t.Fatal("crashed acquisition became a permit")
710 }
711 q.Operation = "executionSettle"
712 q.ID = "settle"
713 q.Token = record.Token
714 if _, err = b.Execute(t.Context(), p, q); err == nil {
715 t.Fatal("uncertain unexpired acquisition settled")
716 }
717 clock.advance(TTL + time.Second)
718 if _, err = b.Execute(t.Context(), p, q); err != nil {
719 t.Fatal("expired acquired obligation could not settle", err)
720 }
721}
722
723func TestPlan14ExecutionRenewCannotCrossPriorCutoff(t *testing.T) {
724 b, _, remote, clock, q, peer := brokerFixture(t)
725 out, err := b.Execute(t.Context(), peer, q)
726 if err != nil {
727 t.Fatal(err)
728 }
729 clock.advance(80 * time.Second)
730 remote.onPut = func(int) { clock.advance(6 * time.Second) }
731 q.Operation, q.ID, q.Token = "executionRenew", "late-renew", out.Token
732 if out, err := b.Execute(t.Context(), peer, q); err == nil {
733 t.Fatalf("late renewal revived expired admission: %+v", out)
734 }
735}
736