1package daemon
2
3import (
4 "context"
5 "io"
6 "sync"
7 "sync/atomic"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/mock"
12 "github.com/stretchr/testify/require"
13
14 "github.com/bnema/vev/internal/domain"
15 "github.com/bnema/vev/internal/ports"
16 portsmocks "github.com/bnema/vev/internal/ports/mocks"
17 "github.com/bnema/vev/internal/protocol"
18 "github.com/bnema/vev/internal/protocol/wire"
19)
20
21func TestSnapshotCoordinatorQuarantineCancelsPublicationBeforeDelete(t *testing.T) {
22 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
23 repository := portsmocks.NewMockSnapshotRepository(t)
24 WithSnapshotRepository(repository)(d)
25 startSnapshotEncodeWorker(t, d)
26
27 started := make(chan struct{})
28 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(ctx context.Context, _ ports.SnapshotPublication) error {
29 close(started)
30 <-ctx.Done()
31 return ctx.Err()
32 }).Once()
33 sess := newSnapshotTestSession(t, "work", false, "/work")
34 markSnapshotDirty(sess)
35 require.True(t, d.scheduleSnapshot(sess))
36 select {
37 case <-started:
38 case <-time.After(testWaitTimeout):
39 t.Fatal("publication did not start")
40 }
41
42 done := quarantineSnapshotCoordinator(sess)
43 select {
44 case <-done:
45 case <-time.After(testWaitTimeout):
46 t.Fatal("quarantine did not join canceled publication")
47 }
48}
49
50func TestSnapshotCoordinatorQuarantineJoinsInFlightPublication(t *testing.T) {
51 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
52 repository := portsmocks.NewMockSnapshotRepository(t)
53 WithSnapshotRepository(repository)(d)
54 startSnapshotEncodeWorker(t, d)
55
56 started := make(chan struct{})
57 release := make(chan struct{})
58 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(context.Context, ports.SnapshotPublication) error {
59 close(started)
60 <-release // deliberately ignore cancellation
61 return nil
62 }).Once()
63 sess := newSnapshotTestSession(t, "work", false, "/work")
64 markSnapshotDirty(sess)
65 require.True(t, d.scheduleSnapshot(sess))
66 select {
67 case <-started:
68 case <-time.After(testWaitTimeout):
69 t.Fatal("publication did not start")
70 }
71
72 done := quarantineSnapshotCoordinator(sess)
73 select {
74 case <-done:
75 t.Fatal("quarantine joined before publication returned")
76 default:
77 }
78 close(release)
79 select {
80 case <-done:
81 case <-time.After(testWaitTimeout):
82 t.Fatal("quarantine did not join publication")
83 }
84}
85
86// TestQuarantineDiscardDoesNotDoubleDecrementPendingCaptures proves the pending
87// capture accounting is exact across a quarantine discard and a rollback
88// replacement: a queued capture the quarantine discarded already dropped its own
89// pending count, so the worker-side finish that rejects its stale identity must
90// not decrement a replacement scheduled after the rollback.
91func TestQuarantineDiscardDoesNotDoubleDecrementPendingCaptures(t *testing.T) {
92 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
93 repository := portsmocks.NewMockSnapshotRepository(t)
94 WithSnapshotRepository(repository)(d)
95 startSnapshotEncodeWorker(t, d)
96
97 blockerStarted := make(chan struct{})
98 releaseBlocker := make(chan struct{})
99 var releaseOnce sync.Once
100 release := func() { releaseOnce.Do(func() { close(releaseBlocker) }) }
101 t.Cleanup(release)
102 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
103 if publication.Name == "blocker" {
104 close(blockerStarted)
105 <-releaseBlocker
106 }
107 return nil
108 }).Maybe()
109
110 // Occupy the single worker so the test session's capture stays queued.
111 blocker := newSnapshotTestSession(t, "blocker", false, "/work")
112 markSnapshotDirty(blocker)
113 require.True(t, d.scheduleSnapshot(blocker))
114 awaitTestValue(t, blockerStarted, "the blocking publication did not start")
115
116 sess := newSnapshotTestSession(t, "work", false, "/work")
117 markSnapshotDirty(sess)
118 require.True(t, d.scheduleSnapshot(sess), "the test capture should queue behind the blocked worker")
119 sess.snapshotMu.Lock()
120 first := sess.snapshotQueuedCapture
121 firstCount := sess.snapshotPendingCaptures
122 sess.snapshotMu.Unlock()
123 require.NotNil(t, first)
124 require.Equal(t, uint(1), firstCount)
125
126 // Quarantine discards the queued capture and drops its pending count exactly
127 // once.
128 <-quarantineSnapshotCoordinator(sess)
129 sess.snapshotMu.Lock()
130 discardedCount := sess.snapshotPendingCaptures
131 sess.snapshotMu.Unlock()
132 require.Equal(t, uint(0), discardedCount, "quarantine must consume the discarded capture's count")
133 require.True(t, first.coordinatorDiscarded)
134
135 // Model the worker having already dequeued the stale capture: the global
136 // bounded queue still holds it, so free the slot before the replacement.
137 select {
138 case dequeued := <-d.snapshotJobs:
139 require.Same(t, first, dequeued)
140 default:
141 t.Fatal("the discarded capture must still be queued in the global worker queue")
142 }
143
144 // The rollback restores scheduling, and the replacement capture owns its own
145 // pending count.
146 rollbackSnapshotQuarantine(sess, true)
147 require.True(t, d.scheduleSnapshot(sess), "the rolled-back session must schedule a replacement capture")
148 sess.snapshotMu.Lock()
149 replacement := sess.snapshotQueuedCapture
150 replacementCount := sess.snapshotPendingCaptures
151 sess.snapshotMu.Unlock()
152 require.NotNil(t, replacement)
153 require.NotSame(t, first, replacement)
154 require.Equal(t, uint(1), replacementCount)
155
156 // The worker finishing the stale discarded capture must not decrement the
157 // live replacement's pending count.
158 d.finishSnapshotCapture(first, false)
159 sess.snapshotMu.Lock()
160 afterFinishCount := sess.snapshotPendingCaptures
161 afterFinishPending := sess.snapshotPending
162 sess.snapshotMu.Unlock()
163 require.Equal(t, uint(1), afterFinishCount, "the discarded capture must not decrement the replacement")
164 require.True(t, afterFinishPending)
165
166 // Once the worker drains both captures only the replacement publishes, and
167 // the accounting returns to zero.
168 release()
169 awaitSnapshotClean(t, sess)
170 sess.snapshotMu.Lock()
171 finalCount := sess.snapshotPendingCaptures
172 finalPending := sess.snapshotPending
173 sess.snapshotMu.Unlock()
174 require.Equal(t, uint(0), finalCount)
175 require.False(t, finalPending)
176}
177
178func TestForcedSnapshotSchedulesSuccessorAfterRoutineInFlight(t *testing.T) {
179 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
180 repository := portsmocks.NewMockSnapshotRepository(t)
181 WithSnapshotRepository(repository)(d)
182 startSnapshotEncodeWorker(t, d)
183
184 started := make(chan struct{})
185 release := make(chan struct{})
186 publications := make(chan ports.SnapshotPublication, 2)
187 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
188 publications <- publication
189 if publication.Generation == 1 {
190 close(started)
191 <-release
192 }
193 return nil
194 }).Twice()
195
196 sess := newSnapshotTestSession(t, "work", false, "/work")
197 markSnapshotDirty(sess)
198 require.True(t, d.scheduleSnapshot(sess))
199 select {
200 case <-started:
201 case <-time.After(testWaitTimeout):
202 t.Fatal("routine publication did not start")
203 }
204
205 // The terminal state changed after the routine capture, so its publication
206 // cannot satisfy the forced checkpoint.
207 sess.tabs[0].panes["pane-1"].screen.Write([]byte(" terminal"))
208 markSnapshotDirty(sess)
209 require.True(t, d.scheduleFinalSnapshot(sess))
210 close(release)
211
212 awaitSnapshotClean(t, sess)
213 require.Equal(t, uint64(1), (<-publications).Generation)
214 require.Equal(t, uint64(2), (<-publications).Generation)
215}
216
217func TestForcedSnapshotSchedulesSuccessorForRoutineCaptureQueuedBehindWorker(t *testing.T) {
218 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
219 repository := portsmocks.NewMockSnapshotRepository(t)
220 WithSnapshotRepository(repository)(d)
221 startSnapshotEncodeWorker(t, d)
222
223 blockerStarted := make(chan struct{})
224 releaseBlocker := make(chan struct{})
225 workPublications := make(chan ports.SnapshotPublication, 2)
226 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
227 if publication.Name == "blocker" {
228 close(blockerStarted)
229 <-releaseBlocker
230 return nil
231 }
232 workPublications <- publication
233 return nil
234 }).Times(3)
235
236 blocker := newSnapshotTestSession(t, "blocker", false, "/work")
237 markSnapshotDirty(blocker)
238 require.True(t, d.scheduleSnapshot(blocker))
239 select {
240 case <-blockerStarted:
241 case <-time.After(testWaitTimeout):
242 t.Fatal("blocking publication did not start")
243 }
244
245 sess := newSnapshotTestSession(t, "work", false, "/work")
246 markSnapshotDirty(sess)
247 require.True(t, d.scheduleSnapshot(sess), "routine capture should occupy the bounded queue")
248 sess.tabs[0].panes["pane-1"].screen.Write([]byte(" terminal"))
249 markSnapshotDirty(sess)
250 require.True(t, d.scheduleFinalSnapshot(sess))
251 close(releaseBlocker)
252
253 awaitSnapshotClean(t, sess)
254 require.Equal(t, uint64(1), (<-workPublications).Generation)
255 require.Equal(t, uint64(2), (<-workPublications).Generation)
256}
257
258func TestMultipleForcedSnapshotsCoalesceToOneSuccessor(t *testing.T) {
259 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
260 repository := portsmocks.NewMockSnapshotRepository(t)
261 WithSnapshotRepository(repository)(d)
262 startSnapshotEncodeWorker(t, d)
263
264 started := make(chan struct{})
265 release := make(chan struct{})
266 publications := make(chan ports.SnapshotPublication, 2)
267 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
268 publications <- publication
269 if publication.Generation == 1 {
270 close(started)
271 <-release
272 }
273 return nil
274 }).Twice()
275
276 sess := newSnapshotTestSession(t, "work", false, "/work")
277 markSnapshotDirty(sess)
278 require.True(t, d.scheduleSnapshot(sess))
279 select {
280 case <-started:
281 case <-time.After(testWaitTimeout):
282 t.Fatal("routine publication did not start")
283 }
284 markSnapshotDirty(sess)
285 require.True(t, d.scheduleFinalSnapshot(sess))
286 markSnapshotDirty(sess)
287 require.True(t, d.scheduleFinalSnapshot(sess))
288 close(release)
289
290 awaitSnapshotClean(t, sess)
291 require.Equal(t, uint64(1), (<-publications).Generation)
292 require.Equal(t, uint64(2), (<-publications).Generation)
293 sess.snapshotMu.Lock()
294 require.False(t, sess.snapshotPending)
295 require.Nil(t, sess.snapshotQueuedCapture)
296 require.Nil(t, sess.snapshotInFlightCapture)
297 sess.snapshotMu.Unlock()
298}
299
300func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.T) {
301 clock := newSnapshotDeadlineClock()
302 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
303 repository := portsmocks.NewMockSnapshotRepository(t)
304 WithSnapshotRepository(repository)(d)
305 notices := portsmocks.NewMockNoticeStore(t)
306 WithNoticeStore(notices)(d)
307 startSnapshotEncodeWorker(t, d)
308
309 blockerStarted := make(chan struct{})
310 releaseBlocker := make(chan struct{})
311 var releaseBlockerOnce sync.Once
312 release := func() { releaseBlockerOnce.Do(func() { close(releaseBlocker) }) }
313 // Register this after newTestDaemon's worker cleanup so the uncooperative
314 // repository call is always released before cleanup asks the worker to join.
315 t.Cleanup(release)
316 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
317 require.Equal(t, "blocker", publication.Name)
318 close(blockerStarted)
319 <-releaseBlocker
320 return nil
321 }).Once()
322 notices.EXPECT().Append(mock.MatchedBy(func(notification domain.Notification) bool {
323 return notification.Code == domain.NoticeSnapshotWrite && notification.Severity == domain.NoticeError
324 })).Return(nil).Once()
325
326 blocker := newSnapshotTestSession(t, "blocker", false, "/work")
327 markSnapshotDirty(blocker)
328 require.True(t, d.scheduleSnapshot(blocker))
329 select {
330 case <-blockerStarted:
331 case <-time.After(testWaitTimeout):
332 t.Fatal("blocking publication did not start")
333 }
334
335 sess := newSnapshotTestSession(t, "work", false, "/work")
336 pty, ok := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
337 require.True(t, ok, "snapshot test pane must use MockPTY")
338 pty.EXPECT().Close().Return(nil).Maybe()
339 d.sessions[sess.id] = sess
340 markSnapshotDirty(sess)
341 require.True(t, d.scheduleSnapshot(sess), "routine capture should occupy the bounded queue")
342
343 killed := make(chan error, 1)
344 go func() { killed <- d.killSession(sess, protocol.ReasonServerShutdown, false) }()
345 timer := clock.nextTimer(t)
346 timer.fire()
347 require.Error(t, <-killed)
348
349 func() {
350 sess.snapshotMu.Lock()
351 defer sess.snapshotMu.Unlock()
352 require.True(t, sess.snapDirty.Load())
353 require.Greater(t, sess.snapshotForcedGeneration, sess.snapshotPublishedGeneration)
354 require.True(t, sess.snapshotPending)
355 require.Equal(t, uint(1), sess.snapshotPendingCaptures)
356 require.NotNil(t, sess.snapshotQueuedCapture, "the one routine capture remains retryable")
357 require.Nil(t, sess.snapshotInFlightCapture)
358 }()
359 d.snapshotWorkerMu.Lock()
360 require.LessOrEqual(t, len(d.snapshotJobs), snapshotQueueCapacity)
361 d.snapshotWorkerMu.Unlock()
362
363 // The retained capture is not published after quarantine, but releasing the
364 // unrelated blocked call lets the worker discard it and lets cleanup join.
365 release()
366 for {
367 sess.snapshotMu.Lock()
368 complete := !sess.snapshotPending && sess.snapshotQueuedCapture == nil
369 changed := sess.snapshotChangeLocked()
370 sess.snapshotMu.Unlock()
371 if complete {
372 break
373 }
374 select {
375 case <-changed:
376 case <-time.After(testWaitTimeout):
377 t.Fatal("quarantined capture was not discarded")
378 }
379 }
380}
381
382func TestServeShutdownDeadlineStillJoinsRetainedSessionWriter(t *testing.T) {
383 pty, releasePTY := newBlockingPTY(t)
384 defer releasePTY()
385 clock := newServeShutdownClock()
386 d := newTestDaemon(t, newFactory(t, pty), clock)
387 repository := portsmocks.NewMockSnapshotRepository(t)
388 WithSnapshotRepository(repository)(d)
389
390 publicationStarted := make(chan struct{})
391 releasePublication := make(chan struct{})
392 var releasePublicationOnce sync.Once
393 releaseSnapshot := func() { releasePublicationOnce.Do(func() { close(releasePublication) }) }
394 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
395 require.Equal(t, uint64(1), publication.Generation)
396 close(publicationStarted)
397 <-releasePublication
398 return nil
399 }).Once()
400
401 tr, sends, _ := newConn(t, mustHello(protocol.IntentNew, "work", domain.Size{Cols: 80, Rows: 24}))
402 listener := serveSnapshotListener(t, tr)
403 ctx, cancel := context.WithCancel(t.Context())
404 defer cancel()
405 served := make(chan error, 1)
406 go func() { served <- d.Serve(ctx, listener) }()
407
408 awaitFrame(t, sends, "Welcome")
409 sess := firstSession(d)
410 require.NotNil(t, sess)
411
412 // A retained-session teardown still publishes a terminal snapshot. It must
413 // own teardown before Serve's context cancellation enters its competing
414 // shutdownAll path.
415 go func() { _ = d.killSession(sess, protocol.ReasonServerShutdown, false) }()
416 select {
417 case <-publicationStarted:
418 case <-time.After(testWaitTimeout):
419 t.Fatal("pane EOF teardown did not publish its terminal snapshot")
420 }
421 // The pane-exit owner has an independent final-snapshot timer. Keep it
422 // unfired until cleanup so Serve must use, and remain bounded by, its shared
423 // deadline.
424 ownerDeadlineTimer := clock.nextFinalTimer(t)
425
426 // Cleanup fires the pane-exit owner's independent budget and joins the real
427 // pane reader. Serve itself must remain owned until the repository writer is
428 // released below, even after its checkpoint deadline expires.
429 t.Cleanup(func() {
430 releaseSnapshot()
431 ownerDeadlineTimer.fire()
432 d.sessWg.Wait()
433 })
434
435 cancel()
436 shutdownDeadlineTimer := clock.nextFinalTimer(t)
437 shutdownDeadlineTimer.fire()
438
439 select {
440 case err := <-served:
441 t.Fatalf("Serve returned with a durable writer alive: %v", err)
442 default:
443 }
444 releaseSnapshot()
445 select {
446 case err := <-served:
447 require.NoError(t, err)
448 case <-time.After(testWaitTimeout):
449 t.Fatal("Serve did not return after the durable writer exited")
450 }
451
452 sess.snapshotMu.Lock()
453 generation := sess.snapshotGeneration
454 sess.snapshotMu.Unlock()
455 require.Equal(t, uint64(1), generation, "Serve shutdown must not create a competing terminal generation")
456 require.Equal(t, int64(snapshotFinalFlushTimeout), clock.finalBudget.Load(), "Serve must spend only its shared snapshot deadline")
457}
458
459func awaitTeardownWaiters(t *testing.T, sess *session, want uint) {
460 t.Helper()
461 for {
462 sess.teardownMu.Lock()
463 waiters := sess.teardownWaiters
464 changed := sess.teardownChangeLocked()
465 sess.teardownMu.Unlock()
466 if waiters == want {
467 return
468 }
469 select {
470 case <-changed:
471 case <-time.After(testWaitTimeout):
472 t.Fatalf("teardown ownership waiters = %d, want %d", waiters, want)
473 }
474 }
475}
476
477func TestConcurrentSessionTeardownPublishesOneFinalSnapshot(t *testing.T) {
478 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{})
479 repository := portsmocks.NewMockSnapshotRepository(t)
480 WithSnapshotRepository(repository)(d)
481 startSnapshotEncodeWorker(t, d)
482
483 publicationStarted := make(chan struct{})
484 releasePublication := make(chan struct{})
485 var releaseOnce sync.Once
486 t.Cleanup(func() { releaseOnce.Do(func() { close(releasePublication) }) })
487 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, _ ports.SnapshotPublication) error {
488 close(publicationStarted)
489 <-releasePublication
490 return nil
491 }).Once()
492
493 sess := newSnapshotTestSession(t, "work", false, "/work")
494 pty, ok := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY)
495 require.True(t, ok)
496 pty.EXPECT().Close().Return(nil).Maybe()
497 d.sessions[sess.id] = sess
498
499 first := make(chan error, 1)
500 go func() { first <- d.killSession(sess, protocol.ReasonSessionKilled, false) }()
501 select {
502 case <-publicationStarted:
503 case <-time.After(testWaitTimeout):
504 t.Fatal("ordinary teardown did not publish its terminal snapshot")
505 }
506
507 sess.teardownMu.Lock()
508 require.True(t, sess.teardownActive)
509 require.NotNil(t, sess.teardownDone)
510 sess.teardownMu.Unlock()
511
512 const duplicateCallers = 8
513 duplicates := make(chan error, duplicateCallers)
514 for range duplicateCallers {
515 go func() {
516 duplicates <- d.killSession(sess, protocol.ReasonSessionKilled, false)
517 }()
518 }
519 // Synchronize on the ownership state itself: every duplicate is waiting
520 // behind the distinct ordinary teardown before its publication is released.
521 awaitTeardownWaiters(t, sess, duplicateCallers)
522
523 sess.snapshotMu.Lock()
524 generation := sess.snapshotGeneration
525 sess.snapshotMu.Unlock()
526 require.Equal(t, uint64(1), generation, "duplicate teardown callers must not create synthetic terminal generations")
527
528 releaseOnce.Do(func() { close(releasePublication) })
529 select {
530 case err := <-first:
531 require.NoError(t, err)
532 case <-time.After(testWaitTimeout):
533 t.Fatal("ordinary teardown did not finish")
534 }
535 for range duplicateCallers {
536 select {
537 case err := <-duplicates:
538 require.NoError(t, err)
539 case <-time.After(testWaitTimeout):
540 t.Fatal("duplicate teardown did not observe the completed registry transition")
541 }
542 }
543 sess.snapshotMu.Lock()
544 generation = sess.snapshotGeneration
545 sess.snapshotMu.Unlock()
546 require.Equal(t, uint64(1), generation, "completed duplicate teardowns must leave the terminal generation unchanged")
547}
548
549func TestServeShutdownCheckpointsBeforeStoppingSnapshotWorker(t *testing.T) {
550 p, releasePTY := newBlockingPTY(t)
551 defer releasePTY()
552 d := newTestDaemon(t, newFactory(t, p), stubClock{})
553 repository := portsmocks.NewMockSnapshotRepository(t)
554 WithSnapshotRepository(repository)(d)
555
556 published := make(chan ports.SnapshotPublication, 2)
557 allowTerminalPublication := make(chan struct{})
558 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
559 published <- publication
560 if publication.Generation == 2 {
561 <-allowTerminalPublication
562 }
563 return nil
564 }).Twice()
565
566 tr, sends, _ := newConn(t, mustHello(protocol.IntentNew, "work", domain.Size{Cols: 80, Rows: 24}))
567 l := serveSnapshotListener(t, tr)
568 ctx, cancel := context.WithCancel(t.Context())
569 defer cancel()
570 served := make(chan error, 1)
571 go func() { served <- d.Serve(ctx, l) }()
572
573 awaitFrame(t, sends, "Welcome")
574 markSnapshotDirty(firstSession(d))
575 d.snapshotWake <- struct{}{}
576 first := <-published
577 require.Equal(t, uint64(1), first.Generation)
578
579 cancel()
580 terminal := <-published
581 require.Equal(t, uint64(2), terminal.Generation, "shutdown must publish a forced terminal checkpoint")
582 close(allowTerminalPublication)
583
584 select {
585 case err := <-served:
586 require.NoError(t, err)
587 case <-time.After(testWaitTimeout):
588 t.Fatal("Serve did not return after the terminal checkpoint completed")
589 }
590}
591
592func TestServeShutdownDeadlineStillJoinsUncooperativeSnapshotRepository(t *testing.T) {
593 p, releasePTY := newBlockingPTY(t)
594 defer releasePTY()
595 clock := newServeShutdownClock()
596 d := newTestDaemon(t, newFactory(t, p), clock)
597 repository := portsmocks.NewMockSnapshotRepository(t)
598 WithSnapshotRepository(repository)(d)
599 noticeStore := portsmocks.NewMockNoticeStore(t)
600 WithNoticeStore(noticeStore)(d)
601 noticeStore.EXPECT().Claim().Return(nil, nil).Maybe()
602 noticeStore.EXPECT().Ack().Return(nil).Maybe()
603 noticePersisted := make(chan struct{})
604 noticeStore.EXPECT().Append(mock.MatchedBy(func(n domain.Notification) bool {
605 return n.Code == domain.NoticeSnapshotWrite && n.Severity == domain.NoticeError
606 })).Run(func(domain.Notification) { close(noticePersisted) }).Return(nil).Once()
607
608 firstPublicationCompleted := make(chan struct{})
609 terminalPublicationStarted := make(chan struct{})
610 releaseTerminalPublication := make(chan struct{})
611 var lastValid atomic.Uint64
612 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
613 if publication.Generation == 1 {
614 lastValid.Store(publication.Generation)
615 close(firstPublicationCompleted)
616 return nil
617 }
618 close(terminalPublicationStarted)
619 <-releaseTerminalPublication // Deliberately ignores the worker cancellation context.
620 return nil
621 }).Twice()
622
623 tr, sends, _ := newConn(t, mustHello(protocol.IntentNew, "work", domain.Size{Cols: 80, Rows: 24}))
624 l := serveSnapshotListener(t, tr)
625 ctx, cancel := context.WithCancel(t.Context())
626 defer cancel()
627 served := make(chan error, 1)
628 go func() { served <- d.Serve(ctx, l) }()
629
630 awaitFrame(t, sends, "Welcome")
631 markSnapshotDirty(firstSession(d))
632 d.snapshotWake <- struct{}{}
633 select {
634 case <-firstPublicationCompleted:
635 case <-time.After(testWaitTimeout):
636 t.Fatal("initial checkpoint did not complete")
637 }
638
639 cancel()
640 select {
641 case <-terminalPublicationStarted:
642 case <-time.After(testWaitTimeout):
643 t.Fatal("shutdown terminal checkpoint did not reach the worker")
644 }
645 d.snapshotWorkerMu.Lock()
646 workerDone := d.snapshotWorkerDone
647 d.snapshotWorkerMu.Unlock()
648 require.NotNil(t, workerDone)
649 clock.nextFinalTimer(t).fire()
650 select {
651 case <-noticePersisted:
652 case <-time.After(testWaitTimeout):
653 t.Fatal("shutdown timeout notice was not persisted before the writer join")
654 }
655
656 select {
657 case err := <-served:
658 t.Fatalf("Serve returned with an uncooperative durable writer alive: %v", err)
659 default:
660 }
661 close(releaseTerminalPublication)
662 select {
663 case <-workerDone:
664 case <-time.After(testWaitTimeout):
665 t.Fatal("snapshot worker leaked after its repository call was released")
666 }
667 select {
668 case err := <-served:
669 require.NoError(t, err)
670 case <-time.After(testWaitTimeout):
671 t.Fatal("Serve did not return after the uncooperative writer exited")
672 }
673 require.LessOrEqual(t, clock.finalBudget.Load(), int64(snapshotFinalFlushTimeout), "shutdown must consume at most one snapshot deadline")
674 require.Equal(t, uint64(1), lastValid.Load(), "the last valid checkpoint remains available after the terminal attempt times out")
675}
676
677type serveShutdownClock struct {
678 finalTimers chan *snapshotDeadlineTimer
679 finalBudget atomic.Int64
680}
681
682func newServeShutdownClock() *serveShutdownClock {
683 return &serveShutdownClock{finalTimers: make(chan *snapshotDeadlineTimer, 2)}
684}
685
686func (*serveShutdownClock) Now() time.Time { return time.Unix(100, 0) }
687func (c *serveShutdownClock) NewTimer(delay time.Duration) ports.Timer {
688 if delay != snapshotFinalFlushTimeout {
689 return stubTimer{}
690 }
691 timer := &snapshotDeadlineTimer{ch: make(chan time.Time, 1), onFire: func() {
692 c.finalBudget.Add(int64(delay))
693 }}
694 c.finalTimers <- timer
695 return timer
696}
697
698func (c *serveShutdownClock) nextFinalTimer(t *testing.T) *snapshotDeadlineTimer {
699 t.Helper()
700 select {
701 case timer := <-c.finalTimers:
702 return timer
703 case <-time.After(testWaitTimeout):
704 t.Fatal("shutdown checkpoint deadline timer was not created")
705 return nil
706 }
707}
708
709func serveSnapshotListener(t *testing.T, tr wire.Transport) *mockServerListener {
710 t.Helper()
711 l := newMockServerListener(t)
712 connections := make(chan wire.Transport, 1)
713 connections <- tr
714 closed := make(chan struct{})
715 var once sync.Once
716 l.EXPECT().Accept().RunAndReturn(func() (wire.Transport, error) {
717 select {
718 case conn := <-connections:
719 return conn, nil
720 case <-closed:
721 return nil, io.EOF
722 }
723 }).Maybe()
724 l.EXPECT().Close().RunAndReturn(func() error {
725 once.Do(func() { close(closed) })
726 return nil
727 }).Maybe()
728 l.EXPECT().Addr().Return("mock").Maybe()
729 return l
730}
731
732func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) {
733 clock := &snapshotSchedulerClock{now: time.Unix(100, 0)}
734 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
735 repository := portsmocks.NewMockSnapshotRepository(t)
736 WithSnapshotRepository(repository)(d)
737 startSnapshotEncodeWorker(t, d)
738
739 var publishes atomic.Int32
740 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, _ ports.SnapshotPublication) error {
741 publishes.Add(1)
742 return nil
743 }).Times(2)
744
745 sess := newSnapshotTestSession(t, "work", false, "/work")
746 markSnapshotDirty(sess)
747 require.True(t, d.scheduleSnapshot(sess))
748 awaitSnapshotClean(t, sess)
749
750 sess.snapshotMu.Lock()
751 eligibleAt := sess.snapshotNextEligibleAt
752 sess.snapshotMu.Unlock()
753 require.Equal(t, time.Unix(100, 0).Add(snapshotInterval), eligibleAt)
754
755 markSnapshotDirty(sess)
756 require.True(t, d.scheduleSnapshot(sess))
757 require.Equal(t, int32(1), publishes.Load(), "rate-limited routine work must not publish")
758
759 require.True(t, d.scheduleFinalSnapshot(sess))
760 awaitSnapshotClean(t, sess)
761 sess.snapshotMu.Lock()
762 require.Equal(t, eligibleAt, sess.snapshotNextEligibleAt)
763 sess.snapshotMu.Unlock()
764}
765
766type snapshotSchedulerClock struct {
767 mu sync.Mutex
768 now time.Time
769}
770
771func (c *snapshotSchedulerClock) Now() time.Time {
772 c.mu.Lock()
773 defer c.mu.Unlock()
774 return c.now
775}
776
777func (*snapshotSchedulerClock) NewTimer(time.Duration) ports.Timer { return stubTimer{} }
778
779type snapshotDeadlineClock struct {
780 timers chan *snapshotDeadlineTimer
781}
782
783func newSnapshotDeadlineClock() *snapshotDeadlineClock {
784 return &snapshotDeadlineClock{timers: make(chan *snapshotDeadlineTimer, 4)}
785}
786
787func (*snapshotDeadlineClock) Now() time.Time { return time.Unix(100, 0) }
788func (c *snapshotDeadlineClock) NewTimer(time.Duration) ports.Timer {
789 timer := &snapshotDeadlineTimer{ch: make(chan time.Time, 1)}
790 c.timers <- timer
791 return timer
792}
793
794func (c *snapshotDeadlineClock) nextTimer(t *testing.T) *snapshotDeadlineTimer {
795 t.Helper()
796 select {
797 case timer := <-c.timers:
798 return timer
799 case <-time.After(testWaitTimeout):
800 t.Fatal("snapshot deadline timer was not created")
801 return nil
802 }
803}
804
805type snapshotDeadlineTimer struct {
806 ch chan time.Time
807 onFire func()
808}
809
810func (t *snapshotDeadlineTimer) C() <-chan time.Time { return t.ch }
811func (*snapshotDeadlineTimer) Reset(time.Duration) bool { return true }
812func (*snapshotDeadlineTimer) Stop() bool { return true }
813func (t *snapshotDeadlineTimer) fire() {
814 if t.onFire != nil {
815 t.onFire()
816 }
817 t.ch <- time.Unix(101, 0)
818}
819
820// deterministicSnapshotClock records timer resets and advances only when a
821// test explicitly fires its timer. It deliberately has no wall-clock behavior.
822type deterministicSnapshotClock struct {
823 mu sync.Mutex
824 now time.Time
825 timers chan *deterministicSnapshotTimer
826}
827
828func newDeterministicSnapshotClock(now time.Time) *deterministicSnapshotClock {
829 return &deterministicSnapshotClock{now: now, timers: make(chan *deterministicSnapshotTimer, 1)}
830}
831
832func (c *deterministicSnapshotClock) Now() time.Time {
833 c.mu.Lock()
834 defer c.mu.Unlock()
835 return c.now
836}
837
838func (c *deterministicSnapshotClock) advance(now time.Time) {
839 c.mu.Lock()
840 c.now = now
841 c.mu.Unlock()
842}
843
844func (c *deterministicSnapshotClock) NewTimer(time.Duration) ports.Timer {
845 timer := &deterministicSnapshotTimer{clock: c, ch: make(chan time.Time, 1), resets: make(chan time.Duration, 16)}
846 c.timers <- timer
847 return timer
848}
849
850func (c *deterministicSnapshotClock) timer() *deterministicSnapshotTimer { return <-c.timers }
851
852type deterministicSnapshotTimer struct {
853 clock *deterministicSnapshotClock
854 ch chan time.Time
855 resets chan time.Duration
856}
857
858func (t *deterministicSnapshotTimer) C() <-chan time.Time { return t.ch }
859func (t *deterministicSnapshotTimer) Reset(delay time.Duration) bool {
860 t.resets <- delay
861 return true
862}
863func (*deterministicSnapshotTimer) Stop() bool { return true }
864func (t *deterministicSnapshotTimer) fire() { t.ch <- t.clock.Now() }
865
866func awaitSnapshotCleanSignal(sess *session) {
867 for {
868 sess.snapshotMu.Lock()
869 clean := !sess.snapDirty.Load()
870 changed := sess.snapshotChangeLocked()
871 sess.snapshotMu.Unlock()
872 if clean {
873 return
874 }
875 <-changed
876 }
877}
878
879func TestSnapshotCompletionSchedulesRoutineRetriesFromCompletion(t *testing.T) {
880 now := time.Unix(100, 0)
881 clock := newDeterministicSnapshotClock(now)
882 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
883 repository := portsmocks.NewMockSnapshotRepository(t)
884 WithSnapshotRepository(repository)(d)
885
886 for _, tt := range []struct {
887 name string
888 kind snapshotAttemptKind
889 succeeded bool
890 existingEligible time.Time
891 wantDirty bool
892 wantEligible time.Time
893 }{
894 {name: "routine success", kind: snapshotAttemptRoutine, succeeded: true, wantEligible: now.Add(snapshotInterval)},
895 {name: "routine failure", kind: snapshotAttemptRoutine, wantDirty: true, wantEligible: now.Add(snapshotInterval)},
896 {name: "forced completion preserves routine eligibility", kind: snapshotAttemptForced, succeeded: true, existingEligible: now.Add(time.Minute), wantEligible: now.Add(time.Minute)},
897 } {
898 t.Run(tt.name, func(t *testing.T) {
899 sess := newSnapshotTestSession(t, "work", false, "/work")
900 sess.snapshotGeneration = 1
901 sess.snapshotNextEligibleAt = tt.existingEligible
902 sess.snapDirty.Store(true)
903 d.finishSnapshotCapture(&snapshotCapture{session: sess, generation: 1, attemptKind: tt.kind}, tt.succeeded)
904
905 require.Equal(t, tt.wantDirty, sess.snapDirty.Load())
906 sess.snapshotMu.Lock()
907 require.Equal(t, tt.wantEligible, sess.snapshotNextEligibleAt)
908 sess.snapshotMu.Unlock()
909 })
910 }
911}
912
913func TestSnapshotSchedulerImmediateEligibilityAndStaleCapturesRemainDirty(t *testing.T) {
914 now := time.Unix(100, 0)
915 clock := newDeterministicSnapshotClock(now)
916 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
917
918 t.Run("no prior attempt is immediately eligible", func(t *testing.T) {
919 published := make(chan ports.SnapshotPublication, 1)
920 repository := portsmocks.NewMockSnapshotRepository(t)
921 WithSnapshotRepository(repository)(d)
922 startSnapshotEncodeWorker(t, d)
923 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
924 published <- publication
925 return nil
926 }).Once()
927
928 sess := newSnapshotTestSession(t, "immediate", false, "/work")
929 sess.snapshotWake = d.snapshotWake
930 sess.snapDirty.Store(true)
931 d.sessions[sess.id] = sess
932
933 d.scheduleEligibleRepositorySnapshots()
934 select {
935 case publication := <-published:
936 require.Equal(t, "immediate", publication.Name)
937 case <-time.After(testWaitTimeout):
938 t.Fatal("immediately eligible snapshot was not published")
939 }
940 awaitSnapshotClean(t, sess)
941 })
942
943 t.Run("queued and in-flight stale successes cannot clear newer generation", func(t *testing.T) {
944 for _, state := range []struct {
945 name string
946 set func(*session, *snapshotCapture)
947 }{
948 {name: "queued", set: func(sess *session, capture *snapshotCapture) { sess.snapshotQueuedCapture = capture }},
949 {name: "in flight", set: func(sess *session, capture *snapshotCapture) { sess.snapshotInFlightCapture = capture }},
950 } {
951 t.Run(state.name, func(t *testing.T) {
952 sess := newSnapshotTestSession(t, state.name, false, "/work")
953 sess.snapshotGeneration = 1
954 sess.snapDirty.Store(true)
955 capture := &snapshotCapture{session: sess, generation: 1, attemptKind: snapshotAttemptRoutine}
956 sess.snapshotPendingCaptures = 1
957 sess.snapshotPending = true
958 state.set(sess, capture)
959
960 markSnapshotDirty(sess)
961 d.finishSnapshotCapture(capture, true)
962
963 require.True(t, sess.snapDirty.Load())
964 require.Equal(t, uint64(1), sess.snapshotPublishedGeneration)
965 require.Equal(t, now.Add(snapshotInterval), sess.snapshotNextEligibleAt)
966 })
967 }
968 })
969}
970
971func TestSnapshotWorkerQueueIsBoundedAndForcedCapturesCoalesce(t *testing.T) {
972 now := time.Unix(100, 0)
973 clock := newDeterministicSnapshotClock(now)
974 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
975 repository := portsmocks.NewMockSnapshotRepository(t)
976 WithSnapshotRepository(repository)(d)
977 startSnapshotEncodeWorker(t, d)
978
979 started := make(chan struct{})
980 release := make(chan struct{})
981 publications := make(chan ports.SnapshotPublication, 3)
982 var blockFirst sync.Once
983 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
984 publications <- publication
985 if publication.Name == "in-flight" {
986 blockFirst.Do(func() {
987 close(started)
988 <-release
989 })
990 }
991 return nil
992 }).Times(3)
993
994 inFlight := newSnapshotTestSession(t, "in-flight", false, "/work")
995 markSnapshotDirty(inFlight)
996 require.True(t, d.scheduleSnapshot(inFlight))
997 <-started
998
999 queued := newSnapshotTestSession(t, "queued", false, "/work")
1000 markSnapshotDirty(queued)
1001 require.True(t, d.scheduleSnapshot(queued))
1002
1003 d.snapshotWorkerMu.Lock()
1004 require.NotNil(t, d.snapshotWorkerInFlight)
1005 require.Len(t, d.snapshotJobs, snapshotQueueCapacity)
1006 d.snapshotWorkerMu.Unlock()
1007
1008 saturated := newSnapshotTestSession(t, "saturated", false, "/work")
1009 markSnapshotDirty(saturated)
1010 require.False(t, d.scheduleSnapshot(saturated), "a saturated queue must not block the producer")
1011 require.True(t, saturated.snapDirty.Load(), "the rejected capture remains retryable")
1012 saturated.snapshotMu.Lock()
1013 require.Equal(t, now.Add(snapshotInterval), saturated.snapshotNextEligibleAt)
1014 saturated.snapshotMu.Unlock()
1015
1016 // A forced request bypasses the routine rate limit and repeated requests
1017 // coalesce to one successor after the in-flight capture completes.
1018 markSnapshotDirty(inFlight)
1019 require.True(t, d.scheduleFinalSnapshot(inFlight))
1020 markSnapshotDirty(inFlight)
1021 require.True(t, d.scheduleFinalSnapshot(inFlight))
1022 close(release)
1023
1024 gotGenerations := []uint64{(<-publications).Generation, (<-publications).Generation, (<-publications).Generation}
1025 require.ElementsMatch(t, []uint64{1, 1, 2}, gotGenerations, "one forced successor bypasses the routine limit while queued work remains bounded")
1026 awaitSnapshotCleanSignal(inFlight)
1027}
1028
1029func TestSnapshotRepositorySaverUsesEarliestDeadlineAndRecomputes(t *testing.T) {
1030 now := time.Unix(100, 0)
1031 clock := newDeterministicSnapshotClock(now)
1032 d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)
1033 repository := portsmocks.NewMockSnapshotRepository(t)
1034 WithSnapshotRepository(repository)(d)
1035 startSnapshotEncodeWorker(t, d)
1036
1037 published := make(chan ports.SnapshotPublication, 1)
1038 repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
1039 published <- publication
1040 return nil
1041 }).Once()
1042
1043 later := newSnapshotTestSession(t, "later", false, "/work")
1044 later.snapshotWake = d.snapshotWake
1045 later.snapDirty.Store(true)
1046 later.snapshotNextEligibleAt = now.Add(10 * time.Minute)
1047 d.sessions[later.id] = later
1048
1049 ctx, cancel := context.WithCancel(t.Context())
1050 done := make(chan struct{})
1051 go func() {
1052 d.snapshotRepositorySaver(ctx)
1053 close(done)
1054 }()
1055 timer := clock.timer()
1056 require.Equal(t, 10*time.Minute, <-timer.resets)
1057
1058 earlier := newSnapshotTestSession(t, "earlier", false, "/work")
1059 earlier.snapshotWake = d.snapshotWake
1060 earlier.snapDirty.Store(true)
1061 earlier.snapshotNextEligibleAt = now.Add(5 * time.Minute)
1062 d.sessions[earlier.id] = earlier
1063 d.snapshotWake <- struct{}{}
1064 require.Equal(t, 5*time.Minute, <-timer.resets)
1065
1066 clock.advance(now.Add(5 * time.Minute))
1067 timer.fire()
1068 // The saver first recomputes immediately after queueing the due capture.
1069 require.Equal(t, 5*time.Minute, <-timer.resets)
1070 publication := <-published
1071 require.Equal(t, "earlier", publication.Name, "timer fires exactly at the earliest eligibility")
1072 for {
1073 earlier.snapshotMu.Lock()
1074 clean := !earlier.snapDirty.Load()
1075 changed := earlier.snapshotChangeLocked()
1076 earlier.snapshotMu.Unlock()
1077 if clean {
1078 break
1079 }
1080 <-changed
1081 }
1082 // The later dirty session is still the next deadline after completion.
1083 require.Equal(t, 5*time.Minute, <-timer.resets)
1084
1085 delete(d.sessions, later.id)
1086 d.snapshotWake <- struct{}{}
1087 require.Equal(t, 24*time.Hour, <-timer.resets)
1088
1089 cancel()
1090 <-done
1091}