fix(daemon): checkpoint before stopping snapshot workers

commit: a9b46bfd5e71154c2630f133abd0d2ecf3aed7fb
author: brice · 3mo ago · 2026-07-21 23:12 UTC
parent: bd4a6ac
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +200 -4

M internal/usecase/daemon/daemon.go +33 -4


@@ -510,7 +510,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
510	510		// force-close. Finally drain the conn handlers, run one defensive sweep in
511	511		// case any ordering ever leaves a session behind, and join the session
512	512		// goroutines (readers unblock via pty.Close, coordinators via ctx cancel).
513		-	d.shutdownAll(ports.ReasonServerShutdown)
	513	+	checkpointIncomplete := d.shutdownAll(ports.ReasonServerShutdown)
514	514		d.waitNotifies()
515	515		d.hardCancel()
516	516		d.serveCancel()

@@ -519,7 +519,14 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
519	519		// registry snapshot has been removed.
520	520		d.connWg.Wait()
521	521		d.attachmentCleanupWg.Wait()
522		-	d.stopSnapshotEncodeWorker()
	522	+	// Forced terminal checkpoints run before shutdown stops producers or the
	523	+	// worker. If one consumed its bounded deadline, do not spend a second
	524	+	// deadline trying to flush a repository call that already ignored it.
	525	+	if checkpointIncomplete {
	526	+	d.abandonSnapshotEncodeWorker()
	527	+	} else {
	528	+	d.stopSnapshotEncodeWorker()
	529	+	}
523	530		d.shutdownAll(ports.ReasonServerShutdown)
524	531		d.sessWg.Wait()
525	532		d.waitNotifies()

@@ -533,7 +540,7 @@ func (d *Daemon) Serve(ctx context.Context, l ports.Listener) error {
533	540		// closing under the same lock as the snapshot guarantees no session can be
534	541		// inserted after the snapshot: route rejects once closing is set, and both run
535	542		// under d.mu. killSession (which relocks) runs after the lock is released.
536		-	func (d *Daemon) shutdownAll(reason uint8) {
	543	+	func (d *Daemon) shutdownAll(reason uint8) (checkpointIncomplete bool) {
537	544		d.mu.Lock()
538	545		d.closing = true
539	546		for token, parked := range d.parked {

@@ -545,13 +552,35 @@ func (d *Daemon) shutdownAll(reason uint8) {
545	552		d.log.Info("graceful shutdown begin", "reason", reason, "sessions", len(snapshot))
546	553		if empty {
547	554		d.doneOnce.Do(func() { close(d.done) })
548		-	return
	555	+	return false
549	556		}
550	557		for _, s := range snapshot {
551	558		if err := d.killSession(s, reason, false); err != nil {
	559	+	checkpointIncomplete = true
552	560		d.log.Error("closing session with unpersisted terminal state", "err", err)
553	561		}
554	562		}
	563	+	return checkpointIncomplete
	564	+	}
	565	+
	566	+	// abandonSnapshotEncodeWorker stops accepting captures and detaches a worker
	567	+	// whose repository call already exceeded the terminal checkpoint deadline. A
	568	+	// repository implementation is allowed to ignore context cancellation, so
	569	+	// Serve cannot join that call without violating its shutdown bound.
	570	+	func (d *Daemon) abandonSnapshotEncodeWorker() {
	571	+	if d == nil {
	572	+	return
	573	+	}
	574	+	d.snapshotWorkerMu.Lock()
	575	+	cancel := d.snapshotWorkerCancel
	576	+	if cancel == nil \|\| d.snapshotWorkerClosing {
	577	+	d.snapshotWorkerMu.Unlock()
	578	+	return
	579	+	}
	580	+	d.snapshotWorkerClosing = true
	581	+	cancel()
	582	+	d.snapshotWorkerMu.Unlock()
	583	+	d.finishStoppedSnapshotWorker(true)
555	584		}
556	585
557	586		func (d Daemon) sessionsSnapshotLocked() []session {

 	// force-close. Finally drain the conn handlers, run one defensive sweep in
 	// case any ordering ever leaves a session behind, and join the session
 	// goroutines (readers unblock via pty.Close, coordinators via ctx cancel).
-	d.shutdownAll(ports.ReasonServerShutdown)
+	checkpointIncomplete := d.shutdownAll(ports.ReasonServerShutdown)
 	d.waitNotifies()
 	d.hardCancel()
 	d.serveCancel()
 	// registry snapshot has been removed.
 	d.connWg.Wait()
 	d.attachmentCleanupWg.Wait()
-	d.stopSnapshotEncodeWorker()
+	// Forced terminal checkpoints run before shutdown stops producers or the
+	// worker. If one consumed its bounded deadline, do not spend a second
+	// deadline trying to flush a repository call that already ignored it.
+	if checkpointIncomplete {
+		d.abandonSnapshotEncodeWorker()
+	} else {
+		d.stopSnapshotEncodeWorker()
+	}
 	d.shutdownAll(ports.ReasonServerShutdown)
 	d.sessWg.Wait()
 	d.waitNotifies()
 // closing under the same lock as the snapshot guarantees no session can be
 // inserted after the snapshot: route rejects once closing is set, and both run
 // under d.mu. killSession (which relocks) runs after the lock is released.
-func (d *Daemon) shutdownAll(reason uint8) {
+func (d *Daemon) shutdownAll(reason uint8) (checkpointIncomplete bool) {
 	d.mu.Lock()
 	d.closing = true
 	for token, parked := range d.parked {
 	d.log.Info("graceful shutdown begin", "reason", reason, "sessions", len(snapshot))
 	if empty {
 		d.doneOnce.Do(func() { close(d.done) })
-		return
+		return false
 	}
 	for _, s := range snapshot {
 		if err := d.killSession(s, reason, false); err != nil {
+			checkpointIncomplete = true
 			d.log.Error("closing session with unpersisted terminal state", "err", err)
 		}
 	}
+	return checkpointIncomplete
+}
+// abandonSnapshotEncodeWorker stops accepting captures and detaches a worker
+// whose repository call already exceeded the terminal checkpoint deadline. A
+// repository implementation is allowed to ignore context cancellation, so
+// Serve cannot join that call without violating its shutdown bound.
+func (d *Daemon) abandonSnapshotEncodeWorker() {
+	if d == nil {
+		return
+	}
+	d.snapshotWorkerMu.Lock()
+	cancel := d.snapshotWorkerCancel
+	if cancel == nil || d.snapshotWorkerClosing {
+		d.snapshotWorkerMu.Unlock()
+		return
+	}
+	d.snapshotWorkerClosing = true
+	cancel()
+	d.snapshotWorkerMu.Unlock()
+	d.finishStoppedSnapshotWorker(true)
 }
 func (d *Daemon) sessionsSnapshotLocked() []*session {

M internal/usecase/daemon/snapshot_scheduler_test.go +167 -0


@@ -3,6 +3,7 @@ package daemon
3	3		import (
4	4		"context"
5	5		"errors"
	6	+	"io"
6	7		"sync"
7	8		"sync/atomic"
8	9		"testing"

@@ -372,6 +373,172 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.
372	373		d.snapshotWorkerMu.Unlock()
373	374		}
374	375
	376	+	func TestServeShutdownCheckpointsBeforeStoppingSnapshotWorker(t *testing.T) {
	377	+	p, releasePTY := newBlockingPTY(t)
	378	+	defer releasePTY()
	379	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
	380	+	repository := portsmocks.NewMockSnapshotRepository(t)
	381	+	WithSnapshotRepository(repository, nil)(d)
	382	+	repository.EXPECT().List(mock.Anything).Return(nil, nil).Maybe()
	383	+
	384	+	published := make(chan ports.SnapshotPublication, 2)
	385	+	allowTerminalPublication := make(chan struct{})
	386	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
	387	+	published <- publication
	388	+	if publication.Generation == 2 {
	389	+	<-allowTerminalPublication
	390	+	}
	391	+	return nil
	392	+	}).Twice()
	393	+
	394	+	tr, sends, _ := newConn(t, mustHello(ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24}))
	395	+	l := serveSnapshotListener(t, tr)
	396	+	ctx, cancel := context.WithCancel(context.Background())
	397	+	defer cancel()
	398	+	served := make(chan error, 1)
	399	+	go func() { served <- d.Serve(ctx, l) }()
	400	+
	401	+	awaitFrame(t, sends, ports.MsgWelcome)
	402	+	markSnapshotDirty(firstSession(d))
	403	+	d.snapshotWake <- struct{}{}
	404	+	first := <-published
	405	+	require.Equal(t, uint64(1), first.Generation)
	406	+
	407	+	cancel()
	408	+	terminal := <-published
	409	+	require.Equal(t, uint64(2), terminal.Generation, "shutdown must publish a forced terminal checkpoint")
	410	+	close(allowTerminalPublication)
	411	+
	412	+	select {
	413	+	case err := <-served:
	414	+	require.NoError(t, err)
	415	+	case <-time.After(testWaitTimeout):
	416	+	t.Fatal("Serve did not return after the terminal checkpoint completed")
	417	+	}
	418	+	}
	419	+
	420	+	func TestServeShutdownDeadlineDoesNotWaitTwiceForUncooperativeSnapshotRepository(t *testing.T) {
	421	+	p, releasePTY := newBlockingPTY(t)
	422	+	defer releasePTY()
	423	+	clock := newServeShutdownClock()
	424	+	d := newTestDaemon(t, newFactory(t, p), clock)
	425	+	repository := portsmocks.NewMockSnapshotRepository(t)
	426	+	WithSnapshotRepository(repository, nil)(d)
	427	+	repository.EXPECT().List(mock.Anything).Return(nil, nil).Maybe()
	428	+	noticeStore := portsmocks.NewMockNoticeStore(t)
	429	+	WithNoticeStore(noticeStore)(d)
	430	+	noticeStore.EXPECT().Claim().Return(nil, nil).Maybe()
	431	+	noticeStore.EXPECT().Ack().Return(nil).Maybe()
	432	+	noticeStore.EXPECT().Append(mock.MatchedBy(func(n domain.Notification) bool {
	433	+	return n.Code == domain.NoticeSnapshotWrite && n.Severity == domain.NoticeError
	434	+	})).Return(nil).Once()
	435	+
	436	+	terminalPublicationStarted := make(chan struct{})
	437	+	releaseTerminalPublication := make(chan struct{})
	438	+	var lastValid atomic.Uint64
	439	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
	440	+	if publication.Generation == 1 {
	441	+	lastValid.Store(publication.Generation)
	442	+	return nil
	443	+	}
	444	+	close(terminalPublicationStarted)
	445	+	<-releaseTerminalPublication // Deliberately ignores the worker cancellation context.
	446	+	return nil
	447	+	}).Twice()
	448	+
	449	+	tr, sends, _ := newConn(t, mustHello(ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24}))
	450	+	l := serveSnapshotListener(t, tr)
	451	+	ctx, cancel := context.WithCancel(context.Background())
	452	+	defer cancel()
	453	+	served := make(chan error, 1)
	454	+	go func() { served <- d.Serve(ctx, l) }()
	455	+
	456	+	awaitFrame(t, sends, ports.MsgWelcome)
	457	+	markSnapshotDirty(firstSession(d))
	458	+	d.snapshotWake <- struct{}{}
	459	+	require.Eventually(t, func() bool { return lastValid.Load() == 1 }, testWaitTimeout, time.Millisecond)
	460	+
	461	+	cancel()
	462	+	select {
	463	+	case <-terminalPublicationStarted:
	464	+	case <-time.After(testWaitTimeout):
	465	+	t.Fatal("shutdown terminal checkpoint did not reach the worker")
	466	+	}
	467	+	d.snapshotWorkerMu.Lock()
	468	+	workerDone := d.snapshotWorkerDone
	469	+	d.snapshotWorkerMu.Unlock()
	470	+	require.NotNil(t, workerDone)
	471	+	defer func() {
	472	+	close(releaseTerminalPublication)
	473	+	select {
	474	+	case <-workerDone:
	475	+	case <-time.After(testWaitTimeout):
	476	+	t.Error("snapshot worker leaked after its repository call was released")
	477	+	}
	478	+	}()
	479	+	clock.nextFinalTimer(t).fire()
	480	+
	481	+	select {
	482	+	case err := <-served:
	483	+	require.NoError(t, err)
	484	+	case <-time.After(testWaitTimeout):
	485	+	t.Fatal("Serve waited beyond the terminal checkpoint deadline")
	486	+	}
	487	+	require.Equal(t, uint64(1), lastValid.Load(), "the last valid checkpoint remains available after the terminal attempt times out")
	488	+	}
	489	+
	490	+	type serveShutdownClock struct {
	491	+	finalTimers chan *snapshotDeadlineTimer
	492	+	}
	493	+
	494	+	func newServeShutdownClock() *serveShutdownClock {
	495	+	return &serveShutdownClock{finalTimers: make(chan *snapshotDeadlineTimer, 2)}
	496	+	}
	497	+
	498	+	func (*serveShutdownClock) Now() time.Time { return time.Unix(100, 0) }
	499	+	func (c *serveShutdownClock) NewTimer(delay time.Duration) ports.Timer {
	500	+	if delay != snapshotFinalFlushTimeout {
	501	+	return stubTimer{}
	502	+	}
	503	+	timer := &snapshotDeadlineTimer{ch: make(chan time.Time, 1)}
	504	+	c.finalTimers <- timer
	505	+	return timer
	506	+	}
	507	+
	508	+	func (c serveShutdownClock) nextFinalTimer(t testing.T) *snapshotDeadlineTimer {
	509	+	t.Helper()
	510	+	select {
	511	+	case timer := <-c.finalTimers:
	512	+	return timer
	513	+	case <-time.After(testWaitTimeout):
	514	+	t.Fatal("shutdown checkpoint deadline timer was not created")
	515	+	return nil
	516	+	}
	517	+	}
	518	+
	519	+	func serveSnapshotListener(t testing.T, tr ports.Transport) portsmocks.MockListener {
	520	+	t.Helper()
	521	+	l := portsmocks.NewMockListener(t)
	522	+	connections := make(chan ports.Transport, 1)
	523	+	connections <- tr
	524	+	closed := make(chan struct{})
	525	+	var once sync.Once
	526	+	l.EXPECT().Accept().RunAndReturn(func() (ports.Transport, error) {
	527	+	select {
	528	+	case conn := <-connections:
	529	+	return conn, nil
	530	+	case <-closed:
	531	+	return nil, io.EOF
	532	+	}
	533	+	}).Maybe()
	534	+	l.EXPECT().Close().RunAndReturn(func() error {
	535	+	once.Do(func() { close(closed) })
	536	+	return nil
	537	+	}).Maybe()
	538	+	l.EXPECT().Addr().Return("mock").Maybe()
	539	+	return l
	540	+	}
	541	+
375	542		func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) {
376	543		clock := &snapshotSchedulerClock{now: time.Unix(100, 0)}
377	544		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)


@@ -3,6 +3,7 @@ package daemon
3	import (	3		import (
4	"context"	4		"context"
5	"errors"	5		"errors"
		6	+	"io"
6	"sync"	7		"sync"
7	"sync/atomic"	8		"sync/atomic"
8	"testing"	9		"testing"

@@ -372,6 +373,172 @@ func TestForcedSnapshotShutdownTimeoutRetainsRetryableStateAndNotice(t *testing.
372	d.snapshotWorkerMu.Unlock()	373		d.snapshotWorkerMu.Unlock()
373	}	374		}
374		375
		376	+	func TestServeShutdownCheckpointsBeforeStoppingSnapshotWorker(t *testing.T) {
		377	+	p, releasePTY := newBlockingPTY(t)
		378	+	defer releasePTY()
		379	+	d := newTestDaemon(t, newFactory(t, p), stubClock{})
		380	+	repository := portsmocks.NewMockSnapshotRepository(t)
		381	+	WithSnapshotRepository(repository, nil)(d)
		382	+	repository.EXPECT().List(mock.Anything).Return(nil, nil).Maybe()
		383	+
		384	+	published := make(chan ports.SnapshotPublication, 2)
		385	+	allowTerminalPublication := make(chan struct{})
		386	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
		387	+	published <- publication
		388	+	if publication.Generation == 2 {
		389	+	<-allowTerminalPublication
		390	+	}
		391	+	return nil
		392	+	}).Twice()
		393	+
		394	+	tr, sends, _ := newConn(t, mustHello(ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24}))
		395	+	l := serveSnapshotListener(t, tr)
		396	+	ctx, cancel := context.WithCancel(context.Background())
		397	+	defer cancel()
		398	+	served := make(chan error, 1)
		399	+	go func() { served <- d.Serve(ctx, l) }()
		400	+
		401	+	awaitFrame(t, sends, ports.MsgWelcome)
		402	+	markSnapshotDirty(firstSession(d))
		403	+	d.snapshotWake <- struct{}{}
		404	+	first := <-published
		405	+	require.Equal(t, uint64(1), first.Generation)
		406	+
		407	+	cancel()
		408	+	terminal := <-published
		409	+	require.Equal(t, uint64(2), terminal.Generation, "shutdown must publish a forced terminal checkpoint")
		410	+	close(allowTerminalPublication)
		411	+
		412	+	select {
		413	+	case err := <-served:
		414	+	require.NoError(t, err)
		415	+	case <-time.After(testWaitTimeout):
		416	+	t.Fatal("Serve did not return after the terminal checkpoint completed")
		417	+	}
		418	+	}
		419	+
		420	+	func TestServeShutdownDeadlineDoesNotWaitTwiceForUncooperativeSnapshotRepository(t *testing.T) {
		421	+	p, releasePTY := newBlockingPTY(t)
		422	+	defer releasePTY()
		423	+	clock := newServeShutdownClock()
		424	+	d := newTestDaemon(t, newFactory(t, p), clock)
		425	+	repository := portsmocks.NewMockSnapshotRepository(t)
		426	+	WithSnapshotRepository(repository, nil)(d)
		427	+	repository.EXPECT().List(mock.Anything).Return(nil, nil).Maybe()
		428	+	noticeStore := portsmocks.NewMockNoticeStore(t)
		429	+	WithNoticeStore(noticeStore)(d)
		430	+	noticeStore.EXPECT().Claim().Return(nil, nil).Maybe()
		431	+	noticeStore.EXPECT().Ack().Return(nil).Maybe()
		432	+	noticeStore.EXPECT().Append(mock.MatchedBy(func(n domain.Notification) bool {
		433	+	return n.Code == domain.NoticeSnapshotWrite && n.Severity == domain.NoticeError
		434	+	})).Return(nil).Once()
		435	+
		436	+	terminalPublicationStarted := make(chan struct{})
		437	+	releaseTerminalPublication := make(chan struct{})
		438	+	var lastValid atomic.Uint64
		439	+	repository.EXPECT().Publish(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, publication ports.SnapshotPublication) error {
		440	+	if publication.Generation == 1 {
		441	+	lastValid.Store(publication.Generation)
		442	+	return nil
		443	+	}
		444	+	close(terminalPublicationStarted)
		445	+	<-releaseTerminalPublication // Deliberately ignores the worker cancellation context.
		446	+	return nil
		447	+	}).Twice()
		448	+
		449	+	tr, sends, _ := newConn(t, mustHello(ports.IntentNew, "work", domain.Size{Cols: 80, Rows: 24}))
		450	+	l := serveSnapshotListener(t, tr)
		451	+	ctx, cancel := context.WithCancel(context.Background())
		452	+	defer cancel()
		453	+	served := make(chan error, 1)
		454	+	go func() { served <- d.Serve(ctx, l) }()
		455	+
		456	+	awaitFrame(t, sends, ports.MsgWelcome)
		457	+	markSnapshotDirty(firstSession(d))
		458	+	d.snapshotWake <- struct{}{}
		459	+	require.Eventually(t, func() bool { return lastValid.Load() == 1 }, testWaitTimeout, time.Millisecond)
		460	+
		461	+	cancel()
		462	+	select {
		463	+	case <-terminalPublicationStarted:
		464	+	case <-time.After(testWaitTimeout):
		465	+	t.Fatal("shutdown terminal checkpoint did not reach the worker")
		466	+	}
		467	+	d.snapshotWorkerMu.Lock()
		468	+	workerDone := d.snapshotWorkerDone
		469	+	d.snapshotWorkerMu.Unlock()
		470	+	require.NotNil(t, workerDone)
		471	+	defer func() {
		472	+	close(releaseTerminalPublication)
		473	+	select {
		474	+	case <-workerDone:
		475	+	case <-time.After(testWaitTimeout):
		476	+	t.Error("snapshot worker leaked after its repository call was released")
		477	+	}
		478	+	}()
		479	+	clock.nextFinalTimer(t).fire()
		480	+
		481	+	select {
		482	+	case err := <-served:
		483	+	require.NoError(t, err)
		484	+	case <-time.After(testWaitTimeout):
		485	+	t.Fatal("Serve waited beyond the terminal checkpoint deadline")
		486	+	}
		487	+	require.Equal(t, uint64(1), lastValid.Load(), "the last valid checkpoint remains available after the terminal attempt times out")
		488	+	}
		489	+
		490	+	type serveShutdownClock struct {
		491	+	finalTimers chan *snapshotDeadlineTimer
		492	+	}
		493	+
		494	+	func newServeShutdownClock() *serveShutdownClock {
		495	+	return &serveShutdownClock{finalTimers: make(chan *snapshotDeadlineTimer, 2)}
		496	+	}
		497	+
		498	+	func (*serveShutdownClock) Now() time.Time { return time.Unix(100, 0) }
		499	+	func (c *serveShutdownClock) NewTimer(delay time.Duration) ports.Timer {
		500	+	if delay != snapshotFinalFlushTimeout {
		501	+	return stubTimer{}
		502	+	}
		503	+	timer := &snapshotDeadlineTimer{ch: make(chan time.Time, 1)}
		504	+	c.finalTimers <- timer
		505	+	return timer
		506	+	}
		507	+
		508	+	func (c serveShutdownClock) nextFinalTimer(t testing.T) *snapshotDeadlineTimer {
		509	+	t.Helper()
		510	+	select {
		511	+	case timer := <-c.finalTimers:
		512	+	return timer
		513	+	case <-time.After(testWaitTimeout):
		514	+	t.Fatal("shutdown checkpoint deadline timer was not created")
		515	+	return nil
		516	+	}
		517	+	}
		518	+
		519	+	func serveSnapshotListener(t testing.T, tr ports.Transport) portsmocks.MockListener {
		520	+	t.Helper()
		521	+	l := portsmocks.NewMockListener(t)
		522	+	connections := make(chan ports.Transport, 1)
		523	+	connections <- tr
		524	+	closed := make(chan struct{})
		525	+	var once sync.Once
		526	+	l.EXPECT().Accept().RunAndReturn(func() (ports.Transport, error) {
		527	+	select {
		528	+	case conn := <-connections:
		529	+	return conn, nil
		530	+	case <-closed:
		531	+	return nil, io.EOF
		532	+	}
		533	+	}).Maybe()
		534	+	l.EXPECT().Close().RunAndReturn(func() error {
		535	+	once.Do(func() { close(closed) })
		536	+	return nil
		537	+	}).Maybe()
		538	+	l.EXPECT().Addr().Return("mock").Maybe()
		539	+	return l
		540	+	}
		541	+
375	func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) {	542		func TestRoutineSnapshotEligibilityStartsAtCompletionAndForcedDoesNotMoveIt(t *testing.T) {
376	clock := &snapshotSchedulerClock{now: time.Unix(100, 0)}	543		clock := &snapshotSchedulerClock{now: time.Unix(100, 0)}
377	d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)	544		d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), clock)