| 1 | package jobs |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "io" |
| 6 | "path/filepath" |
| 7 | "sync" |
| 8 | "sync/atomic" |
| 9 | "testing" |
| 10 | "time" |
| 11 | ) |
| 12 | |
| 13 | func receiveRuntimeState(t *testing.T, states <-chan RuntimeState) RuntimeState { |
| 14 | t.Helper() |
| 15 | select { |
| 16 | case state := <-states: |
| 17 | return state |
| 18 | case <-time.After(5 * time.Second): |
| 19 | t.Fatal("runtime subscriber did not receive a committed state") |
| 20 | return RuntimeState{} |
| 21 | } |
| 22 | } |
| 23 | |
| 24 | // Completion notices are emitted after the drain note is queued, but before |
| 25 | // the job's lifetime ends. Runtime subscribers must observe a separate settled |
| 26 | // lifecycle notification; treating this notice as idle would release guards |
| 27 | // while the job is still unwinding. |
| 28 | func TestRuntimeStateCompletionNoticePrecedesJobExit(t *testing.T) { |
| 29 | sink := &blockingFinishedSink{entered: make(chan struct{}), released: make(chan struct{})} |
| 30 | m := NewManager(sink) |
| 31 | defer m.Close() |
| 32 | m.SetActiveSessionPath("runtime-session", filepath.Join(t.TempDir(), "session.jsonl")) |
| 33 | job := m.StartForSession("runtime-session", "bash", "completion boundary", func(context.Context, io.Writer) (string, error) { |
| 34 | return "isolated result", nil |
| 35 | }) |
| 36 | released := false |
| 37 | defer func() { |
| 38 | if !released { |
| 39 | close(sink.released) |
| 40 | } |
| 41 | }() |
| 42 | select { |
| 43 | case <-sink.entered: |
| 44 | case <-time.After(5 * time.Second): |
| 45 | t.Fatal("completion notice was not delivered") |
| 46 | } |
| 47 | select { |
| 48 | case <-job.done: |
| 49 | t.Fatal("job lifetime ended before completion bookkeeping returned") |
| 50 | default: |
| 51 | } |
| 52 | if got := m.RunningForSession("runtime-session"); len(got) != 1 || got[0].ID != job.ID { |
| 53 | t.Fatalf("notice must not prematurely release runtime protection: %+v", got) |
| 54 | } |
| 55 | if note := m.DrainCompletedNoteForSession("runtime-session"); note == "" { |
| 56 | t.Fatal("completion notice became visible before its drain note") |
| 57 | } |
| 58 | close(sink.released) |
| 59 | released = true |
| 60 | select { |
| 61 | case <-job.done: |
| 62 | case <-time.After(5 * time.Second): |
| 63 | t.Fatal("job failed to unwind after notice delivery") |
| 64 | } |
| 65 | if got := m.RunningForSession("runtime-session"); len(got) != 0 { |
| 66 | t.Fatalf("completed job remains running: %+v", got) |
| 67 | } |
| 68 | } |
| 69 | |
| 70 | func TestRuntimeStateCompletionPublishedAfterJobExit(t *testing.T) { |
| 71 | sink := &blockingFinishedSink{entered: make(chan struct{}), released: make(chan struct{})} |
| 72 | m := NewManager(sink) |
| 73 | defer m.Close() |
| 74 | m.SetActiveSessionPath("runtime-session", filepath.Join(t.TempDir(), "session.jsonl")) |
| 75 | states := make(chan RuntimeState, 4) |
| 76 | initial, unsubscribe := m.SubscribeRuntime("runtime-session", func(state RuntimeState) { states <- state }) |
| 77 | defer unsubscribe() |
| 78 | if initial.Running != 0 || initial.SessionID != "runtime-session" { |
| 79 | t.Fatalf("unexpected initial snapshot: %+v", initial) |
| 80 | } |
| 81 | runRelease := make(chan struct{}) |
| 82 | job := m.StartForSession("runtime-session", "bash", "settled notification", func(context.Context, io.Writer) (string, error) { |
| 83 | <-runRelease |
| 84 | return "done", nil |
| 85 | }) |
| 86 | var releaseOnce sync.Once |
| 87 | defer releaseOnce.Do(func() { close(sink.released) }) |
| 88 | started := receiveRuntimeState(t, states) |
| 89 | close(runRelease) |
| 90 | if started.Running != 1 || started.JobID != job.ID || started.Revision <= initial.Revision { |
| 91 | t.Fatalf("invalid started snapshot: initial=%+v started=%+v", initial, started) |
| 92 | } |
| 93 | select { |
| 94 | case <-sink.entered: |
| 95 | case <-time.After(5 * time.Second): |
| 96 | t.Fatal("completion notice was not delivered") |
| 97 | } |
| 98 | select { |
| 99 | case state := <-states: |
| 100 | t.Fatalf("completion published before job exit: %+v", state) |
| 101 | default: |
| 102 | } |
| 103 | releaseOnce.Do(func() { close(sink.released) }) |
| 104 | completed := receiveRuntimeState(t, states) |
| 105 | if completed.Running != 0 || completed.JobID != job.ID || completed.Revision <= started.Revision { |
| 106 | t.Fatalf("invalid completion snapshot: started=%+v completed=%+v", started, completed) |
| 107 | } |
| 108 | select { |
| 109 | case <-job.done: |
| 110 | default: |
| 111 | t.Fatal("idle snapshot was published before closing the job lifetime") |
| 112 | } |
| 113 | if got := m.RunningForSession("runtime-session"); len(got) != 0 { |
| 114 | t.Fatalf("published idle disagrees with running query: %+v", got) |
| 115 | } |
| 116 | } |
| 117 | |
| 118 | func TestRuntimeStateCancelledJobRetainsProtectionUntilExit(t *testing.T) { |
| 119 | m := NewManager(nil) |
| 120 | defer m.Close() |
| 121 | states := make(chan RuntimeState, 4) |
| 122 | _, unsubscribe := m.SubscribeRuntime("runtime-session", func(state RuntimeState) { states <- state }) |
| 123 | defer unsubscribe() |
| 124 | cancelled := make(chan struct{}) |
| 125 | runRelease := make(chan struct{}) |
| 126 | var releaseOnce sync.Once |
| 127 | defer releaseOnce.Do(func() { close(runRelease) }) |
| 128 | job := m.StartForSession("runtime-session", "bash", "cancel unwind", func(ctx context.Context, _ io.Writer) (string, error) { |
| 129 | <-ctx.Done() |
| 130 | close(cancelled) |
| 131 | <-runRelease |
| 132 | return "", ctx.Err() |
| 133 | }) |
| 134 | started := receiveRuntimeState(t, states) |
| 135 | if !m.KillForSession("runtime-session", job.ID) { |
| 136 | t.Fatal("cancel request was rejected") |
| 137 | } |
| 138 | select { |
| 139 | case <-cancelled: |
| 140 | case <-time.After(5 * time.Second): |
| 141 | t.Fatal("job did not receive cancellation") |
| 142 | } |
| 143 | current, stopProbe := m.SubscribeRuntime("runtime-session", func(RuntimeState) {}) |
| 144 | stopProbe() |
| 145 | if started.Running != 1 || current.Running != 1 { |
| 146 | t.Fatalf("cancellation released protection before exit: started=%+v current=%+v", started, current) |
| 147 | } |
| 148 | releaseOnce.Do(func() { close(runRelease) }) |
| 149 | for { |
| 150 | completed := receiveRuntimeState(t, states) |
| 151 | if completed.Running != 0 { |
| 152 | continue |
| 153 | } |
| 154 | select { |
| 155 | case <-job.done: |
| 156 | default: |
| 157 | t.Fatal("cancelled job published idle before exiting") |
| 158 | } |
| 159 | if completed.Revision <= started.Revision { |
| 160 | t.Fatalf("completion revision did not advance: %+v", completed) |
| 161 | } |
| 162 | break |
| 163 | } |
| 164 | } |
| 165 | |
| 166 | func TestRuntimeStatePublishesForNonActiveSession(t *testing.T) { |
| 167 | m := NewManager(nil) |
| 168 | defer m.Close() |
| 169 | m.SetActiveSession("visible-session") |
| 170 | states := make(chan RuntimeState, 4) |
| 171 | _, unsubscribe := m.SubscribeRuntime("background-session", func(state RuntimeState) { states <- state }) |
| 172 | defer unsubscribe() |
| 173 | runRelease := make(chan struct{}) |
| 174 | job := m.StartForSession("background-session", "bash", "hidden session", func(context.Context, io.Writer) (string, error) { |
| 175 | <-runRelease |
| 176 | return "done", nil |
| 177 | }) |
| 178 | started := receiveRuntimeState(t, states) |
| 179 | close(runRelease) |
| 180 | completed := receiveRuntimeState(t, states) |
| 181 | if started.SessionID != "background-session" || started.Running != 1 || started.JobID != job.ID { |
| 182 | t.Fatalf("non-active start was misrouted: %+v", started) |
| 183 | } |
| 184 | if completed.SessionID != "background-session" || completed.Running != 0 || completed.JobID != job.ID || completed.Revision <= started.Revision { |
| 185 | t.Fatalf("non-active completion was missing or misrouted: %+v", completed) |
| 186 | } |
| 187 | } |
| 188 | |
| 189 | func TestRuntimeStateUnsubscribeDiscardsPendingCallbacks(t *testing.T) { |
| 190 | m := NewManager(nil) |
| 191 | defer m.Close() |
| 192 | entered, release, returned := make(chan struct{}), make(chan struct{}), make(chan struct{}) |
| 193 | var count atomic.Int32 |
| 194 | _, unsubscribe := m.SubscribeRuntime("runtime-session", func(RuntimeState) { |
| 195 | if count.Add(1) == 1 { |
| 196 | close(entered) |
| 197 | <-release |
| 198 | close(returned) |
| 199 | } |
| 200 | }) |
| 201 | defer unsubscribe() |
| 202 | var releaseOnce sync.Once |
| 203 | defer releaseOnce.Do(func() { close(release) }) |
| 204 | job := m.StartForSession("runtime-session", "bash", "unsubscribed job", func(context.Context, io.Writer) (string, error) { return "done", nil }) |
| 205 | select { |
| 206 | case <-entered: |
| 207 | case <-time.After(5 * time.Second): |
| 208 | t.Fatal("subscriber did not enter") |
| 209 | } |
| 210 | select { |
| 211 | case <-job.done: |
| 212 | case <-time.After(5 * time.Second): |
| 213 | t.Fatal("slow subscriber blocked job completion") |
| 214 | } |
| 215 | m.runtimeObservers.mu.Lock() |
| 216 | var subscription *runtimeSubscription |
| 217 | for _, candidate := range m.runtimeObservers.listeners { |
| 218 | subscription = candidate |
| 219 | } |
| 220 | m.runtimeObservers.mu.Unlock() |
| 221 | unsubscribe() |
| 222 | releaseOnce.Do(func() { close(release) }) |
| 223 | <-returned |
| 224 | // Wait for the already-entered callback to leave the dispatcher. This is |
| 225 | // an observation barrier, not a delay used to infer no future callbacks. |
| 226 | waitFor(t, func() bool { |
| 227 | subscription.mu.Lock() |
| 228 | defer subscription.mu.Unlock() |
| 229 | return !subscription.draining |
| 230 | }) |
| 231 | if got := count.Load(); got != 1 { |
| 232 | t.Fatalf("unsubscribe allowed %d callbacks; only the entered callback may finish", got) |
| 233 | } |
| 234 | } |
| 235 |