| 1 | package control |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "strings" |
| 7 | "sync/atomic" |
| 8 | "testing" |
| 9 | "time" |
| 10 | |
| 11 | "reasonix/internal/event" |
| 12 | ) |
| 13 | |
| 14 | // holdFinishingWindow returns a sink that blocks inside the FIRST TurnDone |
| 15 | // delivery until release is closed, holding the controller's finishing window |
| 16 | // deterministically open so tests can place submits inside it. Later |
| 17 | // TurnDones pass through unblocked. |
| 18 | func holdFinishingWindow(release <-chan struct{}, entered chan<- struct{}, events chan<- event.Event) event.Sink { |
| 19 | var first atomic.Int32 |
| 20 | return event.FuncSink(func(e event.Event) { |
| 21 | if e.Kind == event.TurnDone && first.Add(1) == 1 { |
| 22 | entered <- struct{}{} |
| 23 | <-release |
| 24 | } |
| 25 | if events != nil { |
| 26 | select { |
| 27 | case events <- e: |
| 28 | default: |
| 29 | } |
| 30 | } |
| 31 | }) |
| 32 | } |
| 33 | |
| 34 | // TestParkedTurnsRunFIFO pins ordering: several submits landing inside one |
| 35 | // finishing window run in arrival order, one per window close, none lost. |
| 36 | func TestParkedTurnsRunFIFO(t *testing.T) { |
| 37 | entered := make(chan struct{}, 1) |
| 38 | release := make(chan struct{}) |
| 39 | c := newOwnedTestController(t, Options{Sink: holdFinishingWindow(release, entered, nil)}) |
| 40 | |
| 41 | c.runGuarded(func(context.Context) error { return nil }) |
| 42 | <-entered |
| 43 | |
| 44 | var order []int |
| 45 | ran := make(chan int, 3) |
| 46 | for i := 1; i <= 3; i++ { |
| 47 | |
| 48 | if got := c.runGuarded(func(context.Context) error { |
| 49 | ran <- i |
| 50 | return nil |
| 51 | }); got != turnParked { |
| 52 | t.Fatalf("submit %d admission = %v, want turnParked", i, got) |
| 53 | } |
| 54 | } |
| 55 | close(release) |
| 56 | |
| 57 | deadline := time.After(30 * time.Second) |
| 58 | for len(order) < 3 { |
| 59 | select { |
| 60 | case i := <-ran: |
| 61 | order = append(order, i) |
| 62 | case <-deadline: |
| 63 | t.Fatalf("parked turns did not all run; got order %v", order) |
| 64 | } |
| 65 | } |
| 66 | if order[0] != 1 || order[1] != 2 || order[2] != 3 { |
| 67 | t.Fatalf("parked turns ran out of order: %v", order) |
| 68 | } |
| 69 | } |
| 70 | |
| 71 | // TestSubmitDuringRotationEmitsNotice pins the rotating posture: the input's |
| 72 | // intended session is ambiguous while the executor session is being swapped, |
| 73 | // so the submit is refused with a user-visible notice instead of silently |
| 74 | // dropped (the caller should resend against the session they can now see). |
| 75 | func TestSubmitDuringRotationEmitsNotice(t *testing.T) { |
| 76 | events := make(chan event.Event, 8) |
| 77 | c := newOwnedTestController(t, Options{Sink: event.FuncSink(func(e event.Event) { |
| 78 | select { |
| 79 | case events <- e: |
| 80 | default: |
| 81 | } |
| 82 | })}) |
| 83 | |
| 84 | c.mu.Lock() |
| 85 | c.rotating = true |
| 86 | c.mu.Unlock() |
| 87 | |
| 88 | bodyRan := make(chan struct{}, 1) |
| 89 | if got := c.runGuarded(func(context.Context) error { |
| 90 | bodyRan <- struct{}{} |
| 91 | return nil |
| 92 | }); got != turnDroppedRotating { |
| 93 | t.Fatalf("admission during rotation = %v, want turnDroppedRotating", got) |
| 94 | } |
| 95 | |
| 96 | select { |
| 97 | case e := <-events: |
| 98 | if e.Kind != event.Notice || e.Level != event.LevelWarn || !strings.Contains(e.Text, "resend") { |
| 99 | t.Fatalf("event = %+v, want a warn notice asking to resend", e) |
| 100 | } |
| 101 | case <-time.After(time.Second): |
| 102 | t.Fatal("no notice emitted for a rotation-dropped submit") |
| 103 | } |
| 104 | select { |
| 105 | case <-bodyRan: |
| 106 | t.Fatal("rotation-dropped body must not run") |
| 107 | case <-time.After(50 * time.Millisecond): |
| 108 | } |
| 109 | |
| 110 | c.mu.Lock() |
| 111 | c.rotating = false |
| 112 | c.mu.Unlock() |
| 113 | } |
| 114 | |
| 115 | // TestSubmitWhileRunningStaysSilentNoOp pins the running posture: unchanged |
| 116 | // from the historical contract — frontends own the steer/queue UX, internal |
| 117 | // opportunistic callers rely on the quiet no-op. |
| 118 | func TestSubmitWhileRunningStaysSilentNoOp(t *testing.T) { |
| 119 | block := make(chan struct{}) |
| 120 | var notices atomic.Int32 |
| 121 | c := newOwnedTestController(t, Options{Sink: event.FuncSink(func(e event.Event) { |
| 122 | if e.Kind == event.Notice { |
| 123 | notices.Add(1) |
| 124 | } |
| 125 | })}) |
| 126 | |
| 127 | started := make(chan struct{}) |
| 128 | c.runGuarded(func(context.Context) error { |
| 129 | close(started) |
| 130 | <-block |
| 131 | return nil |
| 132 | }) |
| 133 | <-started |
| 134 | |
| 135 | if got := c.runGuarded(func(context.Context) error { return nil }); got != turnDroppedRunning { |
| 136 | t.Fatalf("admission while running = %v, want turnDroppedRunning", got) |
| 137 | } |
| 138 | if n := notices.Load(); n != 0 { |
| 139 | t.Fatalf("running drop should stay silent, got %d notices", n) |
| 140 | } |
| 141 | close(block) |
| 142 | waitIdleAdmission(t, c) |
| 143 | } |
| 144 | |
| 145 | // TestCloseDiscardsParkedTurns pins teardown: a turn parked in the finishing |
| 146 | // window must not start against a controller that has been closed. |
| 147 | func TestCloseDiscardsParkedTurns(t *testing.T) { |
| 148 | entered := make(chan struct{}, 1) |
| 149 | release := make(chan struct{}) |
| 150 | c := newOwnedTestController(t, Options{Sink: holdFinishingWindow(release, entered, nil)}) |
| 151 | |
| 152 | c.runGuarded(func(context.Context) error { return nil }) |
| 153 | <-entered |
| 154 | |
| 155 | parkedRan := make(chan struct{}, 1) |
| 156 | if got := c.runGuarded(func(context.Context) error { |
| 157 | parkedRan <- struct{}{} |
| 158 | return nil |
| 159 | }); got != turnParked { |
| 160 | t.Fatalf("admission = %v, want turnParked", got) |
| 161 | } |
| 162 | |
| 163 | c.Close() |
| 164 | close(release) |
| 165 | |
| 166 | select { |
| 167 | case <-parkedRan: |
| 168 | t.Fatal("parked turn ran after Close") |
| 169 | case <-time.After(200 * time.Millisecond): |
| 170 | } |
| 171 | } |
| 172 | |
| 173 | // TestCloseSealsAdmissionDuringFinishingWindow pins the terminal-state |
| 174 | // ordering the first review round flagged: Close clears the parked queue, but |
| 175 | // a submit arriving AFTER that — while the old turn's TurnDone delivery is |
| 176 | // still in flight — must be rejected outright, not parked and started against |
| 177 | // freed resources when the window closes. |
| 178 | func TestCloseSealsAdmissionDuringFinishingWindow(t *testing.T) { |
| 179 | entered := make(chan struct{}, 1) |
| 180 | release := make(chan struct{}) |
| 181 | c := newOwnedTestController(t, Options{Sink: holdFinishingWindow(release, entered, nil)}) |
| 182 | |
| 183 | c.runGuarded(func(context.Context) error { return nil }) |
| 184 | <-entered // finishing window is now held open |
| 185 | |
| 186 | c.Close() // seals admission; parked queue is empty at this instant |
| 187 | |
| 188 | lateRan := make(chan struct{}, 1) |
| 189 | if got := c.runGuarded(func(context.Context) error { |
| 190 | lateRan <- struct{}{} |
| 191 | return nil |
| 192 | }); got != turnDroppedClosed { |
| 193 | t.Fatalf("submit after Close during finishing window = %v, want turnDroppedClosed", got) |
| 194 | } |
| 195 | |
| 196 | close(release) // window closes; the drain must start nothing |
| 197 | select { |
| 198 | case <-lateRan: |
| 199 | t.Fatal("submit accepted after Close ran when finishing window closed") |
| 200 | case <-time.After(200 * time.Millisecond): |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | // TestRunTurnRefusedDuringFinishingWindow pins the synchronous gate: RunTurn |
| 205 | // must not start inside the previous turn's TurnDone delivery window — that |
| 206 | // would recreate the completion/transport crosstalk the window prevents. |
| 207 | func TestRunTurnRefusedDuringFinishingWindow(t *testing.T) { |
| 208 | entered := make(chan struct{}, 1) |
| 209 | release := make(chan struct{}) |
| 210 | c := newOwnedTestController(t, Options{Sink: holdFinishingWindow(release, entered, nil)}) |
| 211 | |
| 212 | c.runGuarded(func(context.Context) error { return nil }) |
| 213 | <-entered // finishing window is now held open |
| 214 | |
| 215 | errCh := make(chan error, 1) |
| 216 | go func() { errCh <- c.RunTurn(context.Background(), "sync input") }() |
| 217 | select { |
| 218 | case err := <-errCh: |
| 219 | if !errors.Is(err, ErrTurnRunning) { |
| 220 | t.Fatalf("RunTurn during finishing window = %v, want ErrTurnRunning", err) |
| 221 | } |
| 222 | case <-time.After(time.Second): |
| 223 | t.Fatal("RunTurn did not return promptly during the finishing window") |
| 224 | } |
| 225 | close(release) |
| 226 | waitIdleAdmission(t, c) |
| 227 | } |
| 228 | |
| 229 | // TestRunTurnRefusedAfterClose pins the terminal state for the synchronous |
| 230 | // entry point too. |
| 231 | func TestRunTurnRefusedAfterClose(t *testing.T) { |
| 232 | c := newOwnedTestController(t, Options{}) |
| 233 | c.Close() |
| 234 | if err := c.RunTurn(context.Background(), "late"); !errors.Is(err, ErrTurnRunning) { |
| 235 | t.Fatalf("RunTurn after Close = %v, want ErrTurnRunning", err) |
| 236 | } |
| 237 | } |
| 238 | |
| 239 | // waitIdleAdmission waits for the same lifecycle boundary as production turn |
| 240 | // admission, including synchronous TurnDone fan-out. |
| 241 | func waitIdleAdmission(t *testing.T, c *Controller) { |
| 242 | t.Helper() |
| 243 | waitIdle(t, c) |
| 244 | } |
| 245 |