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