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