test(daemon): deterministically assert kill-session snapshot ordering
TestKillSessionSnapshotsNamedSessionBeforeClosingPanes asserted inside the mocked store.Write that panes were already closed, but the encode worker's Write legitimately races the tail of killSession; on a losing schedule the require fired on the worker goroutine, whose Goexit killed the worker before finishSnapshotCapture could signal, and the deadline-less awaitSnapshotClean then hung until the 10-minute package timeout swallowed the real failure. Prove the guarantee without timing dependence instead: gate store.Write, require pane close completed after killSession returns while the gate is still held, then release, wait bounded, and assert the recorded payload on the test goroutine. Give awaitSnapshotClean the same testWaitTimeout deadline as awaitSnapshotIdle so any future missed clean signal fails fast with a clear message.
- commit
- f1556b73f1205b0853454f357ffc2046ed78502a
- author
- brice · · 2026-07-19 06:49 UTC
- parent
- 092c799
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
1 file · +48 -12
M internal/usecase/daemon/snapshot_test.go +48 -12
| @@ -738,24 +738,20 @@ func TestSnapshotCaptureSkipsBareShellProcess(t *testing.T) { | |||
| 738 | 738 | } | |
| 739 | 739 | ||
| 740 | 740 | func TestKillSessionSnapshotsNamedSessionBeforeClosingPanes(t *testing.T) { | |
| 741 | - | store := portsmocks.NewMockSnapshotStore(t) | |
| 741 | + | store := newGatedSnapshotStore() | |
| 742 | 742 | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | |
| 743 | 743 | WithSnapshotStore(store)(d) | |
| 744 | - | startSnapshotEncodeWorker(t, d) | |
| 744 | + | d.startSnapshotEncodeWorker() | |
| 745 | + | t.Cleanup(func() { | |
| 746 | + | store.unblock() | |
| 747 | + | d.stopSnapshotEncodeWorker() | |
| 748 | + | }) | |
| 745 | 749 | d.procCwd = func(int) (string, error) { return "/live", nil } | |
| 746 | 750 | ||
| 747 | 751 | sess := newSnapshotTestSession(t, "work", false, "/fallback") | |
| 748 | 752 | d.sessions[sess.id] = sess | |
| 749 | 753 | ||
| 750 | 754 | var closed atomic.Bool | |
| 751 | - | store.EXPECT().Write("work", mock.Anything).RunAndReturn(func(_ string, data []byte) error { | |
| 752 | - | snap, err := snapcodec.Unmarshal(data) | |
| 753 | - | require.NoError(t, err) | |
| 754 | - | require.Equal(t, "work", snap.Name) | |
| 755 | - | require.Equal(t, "/live", snap.Tabs[0].Panes[0].Cwd) | |
| 756 | - | require.True(t, closed.Load(), "snapshot persistence must not delay pane close") | |
| 757 | - | return nil | |
| 758 | - | }).Once() | |
| 759 | 755 | mockPTY, ok := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY) | |
| 760 | 756 | require.True(t, ok) | |
| 761 | 757 | mockPTY.EXPECT().Close().RunAndReturn(func() error { | |
| @@ -764,7 +760,22 @@ func TestKillSessionSnapshotsNamedSessionBeforeClosingPanes(t *testing.T) { | |||
| 764 | 760 | }).Once() | |
| 765 | 761 | ||
| 766 | 762 | require.NoError(t, d.killSession(sess, 0, false)) | |
| 763 | + | // closeAllPanes runs synchronously inside killSession, which has already | |
| 764 | + | // returned here. The store's write gate is still held below, so | |
| 765 | + | // persistence cannot have completed yet: this proves pane close did not | |
| 766 | + | // wait on it, with zero timing dependence on the encode worker. | |
| 767 | + | require.True(t, closed.Load(), "snapshot persistence must not delay pane close") | |
| 768 | + | ||
| 769 | + | store.unblock() | |
| 767 | 770 | awaitSnapshotClean(t, sess) | |
| 771 | + | ||
| 772 | + | writes, name, data := store.recorded() | |
| 773 | + | require.Equal(t, 1, writes) | |
| 774 | + | require.Equal(t, "work", name) | |
| 775 | + | snap, err := snapcodec.Unmarshal(data) | |
| 776 | + | require.NoError(t, err) | |
| 777 | + | require.Equal(t, "work", snap.Name) | |
| 778 | + | require.Equal(t, "/live", snap.Tabs[0].Panes[0].Cwd) | |
| 768 | 779 | } | |
| 769 | 780 | ||
| 770 | 781 | func TestPTYReaderSynchronizedUpdateFlushMarksSnapshotDirty(t *testing.T) { | |
| @@ -1061,6 +1072,8 @@ func awaitSnapshotIdle(t testing.TB, sess *session) { | |||
| 1061 | 1072 | ||
| 1062 | 1073 | func awaitSnapshotClean(t *testing.T, sess *session) { | |
| 1063 | 1074 | t.Helper() | |
| 1075 | + | timer := time.NewTimer(testWaitTimeout) | |
| 1076 | + | defer timer.Stop() | |
| 1064 | 1077 | for { | |
| 1065 | 1078 | sess.snapshotMu.Lock() | |
| 1066 | 1079 | clean := !sess.snapDirty.Load() | |
| @@ -1069,7 +1082,11 @@ func awaitSnapshotClean(t *testing.T, sess *session) { | |||
| 1069 | 1082 | if clean { | |
| 1070 | 1083 | return | |
| 1071 | 1084 | } | |
| 1072 | - | <-changed | |
| 1085 | + | select { | |
| 1086 | + | case <-changed: | |
| 1087 | + | case <-timer.C: | |
| 1088 | + | t.Fatal("timed out waiting for snapshot to become clean") | |
| 1089 | + | } | |
| 1073 | 1090 | } | |
| 1074 | 1091 | } | |
| 1075 | 1092 | ||
| @@ -1090,6 +1107,11 @@ type gatedSnapshotStore struct { | |||
| 1090 | 1107 | release chan struct{} | |
| 1091 | 1108 | once sync.Once | |
| 1092 | 1109 | releaseOnce sync.Once | |
| 1110 | + | ||
| 1111 | + | mu sync.Mutex | |
| 1112 | + | writes int | |
| 1113 | + | name string | |
| 1114 | + | data []byte | |
| 1093 | 1115 | } | |
| 1094 | 1116 | ||
| 1095 | 1117 | func newGatedSnapshotStore() *gatedSnapshotStore { | |
| @@ -1099,14 +1121,28 @@ func newGatedSnapshotStore() *gatedSnapshotStore { | |||
| 1099 | 1121 | func (s *gatedSnapshotStore) enter() { s.once.Do(func() { close(s.entered) }) } | |
| 1100 | 1122 | func (s *gatedSnapshotStore) unblock() { s.releaseOnce.Do(func() { close(s.release) }) } | |
| 1101 | 1123 | ||
| 1102 | - | func (s *gatedSnapshotStore) Write(_ string, _ []byte) error { | |
| 1124 | + | func (s *gatedSnapshotStore) Write(name string, data []byte) error { | |
| 1103 | 1125 | s.enter() | |
| 1104 | 1126 | <-s.release | |
| 1127 | + | s.mu.Lock() | |
| 1128 | + | s.writes++ | |
| 1129 | + | s.name = name | |
| 1130 | + | s.data = append([]byte(nil), data...) | |
| 1131 | + | s.mu.Unlock() | |
| 1105 | 1132 | return nil | |
| 1106 | 1133 | } | |
| 1107 | 1134 | func (*gatedSnapshotStore) Load() ([]ports.SnapshotBlob, error) { return nil, nil } | |
| 1108 | 1135 | func (*gatedSnapshotStore) Delete(string) error { return nil } | |
| 1109 | 1136 | ||
| 1137 | + | // recorded returns the write count and the most recently written name/data, | |
| 1138 | + | // captured under the store's own lock so it is safe to call from the test | |
| 1139 | + | // goroutine after the write has happened-before via awaitSnapshotClean. | |
| 1140 | + | func (s *gatedSnapshotStore) recorded() (int, string, []byte) { | |
| 1141 | + | s.mu.Lock() | |
| 1142 | + | defer s.mu.Unlock() | |
| 1143 | + | return s.writes, s.name, s.data | |
| 1144 | + | } | |
| 1145 | + | ||
| 1110 | 1146 | type neverReturningSnapshotStore struct { | |
| 1111 | 1147 | entered chan struct{} | |
| 1112 | 1148 | once sync.Once | |
| @@ -738,24 +738,20 @@ func TestSnapshotCaptureSkipsBareShellProcess(t *testing.T) { | |||||
| 738 | } | 738 | } | ||
| 739 | 739 | ||||
| 740 | func TestKillSessionSnapshotsNamedSessionBeforeClosingPanes(t *testing.T) { | 740 | func TestKillSessionSnapshotsNamedSessionBeforeClosingPanes(t *testing.T) { | ||
| 741 | - | store := portsmocks.NewMockSnapshotStore(t) | 741 | + | store := newGatedSnapshotStore() |
| 742 | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | 742 | d := newTestDaemon(t, portsmocks.NewMockPTYFactory(t), stubClock{}) | ||
| 743 | WithSnapshotStore(store)(d) | 743 | WithSnapshotStore(store)(d) | ||
| 744 | - | startSnapshotEncodeWorker(t, d) | 744 | + | d.startSnapshotEncodeWorker() |
| 745 | + | t.Cleanup(func() { | |||
| 746 | + | store.unblock() | |||
| 747 | + | d.stopSnapshotEncodeWorker() | |||
| 748 | + | }) | |||
| 745 | d.procCwd = func(int) (string, error) { return "/live", nil } | 749 | d.procCwd = func(int) (string, error) { return "/live", nil } | ||
| 746 | 750 | ||||
| 747 | sess := newSnapshotTestSession(t, "work", false, "/fallback") | 751 | sess := newSnapshotTestSession(t, "work", false, "/fallback") | ||
| 748 | d.sessions[sess.id] = sess | 752 | d.sessions[sess.id] = sess | ||
| 749 | 753 | ||||
| 750 | var closed atomic.Bool | 754 | var closed atomic.Bool | ||
| 751 | - | store.EXPECT().Write("work", mock.Anything).RunAndReturn(func(_ string, data []byte) error { | |||
| 752 | - | snap, err := snapcodec.Unmarshal(data) | |||
| 753 | - | require.NoError(t, err) | |||
| 754 | - | require.Equal(t, "work", snap.Name) | |||
| 755 | - | require.Equal(t, "/live", snap.Tabs[0].Panes[0].Cwd) | |||
| 756 | - | require.True(t, closed.Load(), "snapshot persistence must not delay pane close") | |||
| 757 | - | return nil | |||
| 758 | - | }).Once() | |||
| 759 | mockPTY, ok := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY) | 755 | mockPTY, ok := sess.tabs[0].panes["pane-1"].pty.(*portsmocks.MockPTY) | ||
| 760 | require.True(t, ok) | 756 | require.True(t, ok) | ||
| 761 | mockPTY.EXPECT().Close().RunAndReturn(func() error { | 757 | mockPTY.EXPECT().Close().RunAndReturn(func() error { | ||
| @@ -764,7 +760,22 @@ func TestKillSessionSnapshotsNamedSessionBeforeClosingPanes(t *testing.T) { | |||||
| 764 | }).Once() | 760 | }).Once() | ||
| 765 | 761 | ||||
| 766 | require.NoError(t, d.killSession(sess, 0, false)) | 762 | require.NoError(t, d.killSession(sess, 0, false)) | ||
| 763 | + | // closeAllPanes runs synchronously inside killSession, which has already | |||
| 764 | + | // returned here. The store's write gate is still held below, so | |||
| 765 | + | // persistence cannot have completed yet: this proves pane close did not | |||
| 766 | + | // wait on it, with zero timing dependence on the encode worker. | |||
| 767 | + | require.True(t, closed.Load(), "snapshot persistence must not delay pane close") | |||
| 768 | + | ||||
| 769 | + | store.unblock() | |||
| 767 | awaitSnapshotClean(t, sess) | 770 | awaitSnapshotClean(t, sess) | ||
| 771 | + | ||||
| 772 | + | writes, name, data := store.recorded() | |||
| 773 | + | require.Equal(t, 1, writes) | |||
| 774 | + | require.Equal(t, "work", name) | |||
| 775 | + | snap, err := snapcodec.Unmarshal(data) | |||
| 776 | + | require.NoError(t, err) | |||
| 777 | + | require.Equal(t, "work", snap.Name) | |||
| 778 | + | require.Equal(t, "/live", snap.Tabs[0].Panes[0].Cwd) | |||
| 768 | } | 779 | } | ||
| 769 | 780 | ||||
| 770 | func TestPTYReaderSynchronizedUpdateFlushMarksSnapshotDirty(t *testing.T) { | 781 | func TestPTYReaderSynchronizedUpdateFlushMarksSnapshotDirty(t *testing.T) { | ||
| @@ -1061,6 +1072,8 @@ func awaitSnapshotIdle(t testing.TB, sess *session) { | |||||
| 1061 | 1072 | ||||
| 1062 | func awaitSnapshotClean(t *testing.T, sess *session) { | 1073 | func awaitSnapshotClean(t *testing.T, sess *session) { | ||
| 1063 | t.Helper() | 1074 | t.Helper() | ||
| 1075 | + | timer := time.NewTimer(testWaitTimeout) | |||
| 1076 | + | defer timer.Stop() | |||
| 1064 | for { | 1077 | for { | ||
| 1065 | sess.snapshotMu.Lock() | 1078 | sess.snapshotMu.Lock() | ||
| 1066 | clean := !sess.snapDirty.Load() | 1079 | clean := !sess.snapDirty.Load() | ||
| @@ -1069,7 +1082,11 @@ func awaitSnapshotClean(t *testing.T, sess *session) { | |||||
| 1069 | if clean { | 1082 | if clean { | ||
| 1070 | return | 1083 | return | ||
| 1071 | } | 1084 | } | ||
| 1072 | - | <-changed | 1085 | + | select { |
| 1086 | + | case <-changed: | |||
| 1087 | + | case <-timer.C: | |||
| 1088 | + | t.Fatal("timed out waiting for snapshot to become clean") | |||
| 1089 | + | } | |||
| 1073 | } | 1090 | } | ||
| 1074 | } | 1091 | } | ||
| 1075 | 1092 | ||||
| @@ -1090,6 +1107,11 @@ type gatedSnapshotStore struct { | |||||
| 1090 | release chan struct{} | 1107 | release chan struct{} | ||
| 1091 | once sync.Once | 1108 | once sync.Once | ||
| 1092 | releaseOnce sync.Once | 1109 | releaseOnce sync.Once | ||
| 1110 | + | ||||
| 1111 | + | mu sync.Mutex | |||
| 1112 | + | writes int | |||
| 1113 | + | name string | |||
| 1114 | + | data []byte | |||
| 1093 | } | 1115 | } | ||
| 1094 | 1116 | ||||
| 1095 | func newGatedSnapshotStore() *gatedSnapshotStore { | 1117 | func newGatedSnapshotStore() *gatedSnapshotStore { | ||
| @@ -1099,14 +1121,28 @@ func newGatedSnapshotStore() *gatedSnapshotStore { | |||||
| 1099 | func (s *gatedSnapshotStore) enter() { s.once.Do(func() { close(s.entered) }) } | 1121 | func (s *gatedSnapshotStore) enter() { s.once.Do(func() { close(s.entered) }) } | ||
| 1100 | func (s *gatedSnapshotStore) unblock() { s.releaseOnce.Do(func() { close(s.release) }) } | 1122 | func (s *gatedSnapshotStore) unblock() { s.releaseOnce.Do(func() { close(s.release) }) } | ||
| 1101 | 1123 | ||||
| 1102 | - | func (s *gatedSnapshotStore) Write(_ string, _ []byte) error { | 1124 | + | func (s *gatedSnapshotStore) Write(name string, data []byte) error { |
| 1103 | s.enter() | 1125 | s.enter() | ||
| 1104 | <-s.release | 1126 | <-s.release | ||
| 1127 | + | s.mu.Lock() | |||
| 1128 | + | s.writes++ | |||
| 1129 | + | s.name = name | |||
| 1130 | + | s.data = append([]byte(nil), data...) | |||
| 1131 | + | s.mu.Unlock() | |||
| 1105 | return nil | 1132 | return nil | ||
| 1106 | } | 1133 | } | ||
| 1107 | func (*gatedSnapshotStore) Load() ([]ports.SnapshotBlob, error) { return nil, nil } | 1134 | func (*gatedSnapshotStore) Load() ([]ports.SnapshotBlob, error) { return nil, nil } | ||
| 1108 | func (*gatedSnapshotStore) Delete(string) error { return nil } | 1135 | func (*gatedSnapshotStore) Delete(string) error { return nil } | ||
| 1109 | 1136 | ||||
| 1137 | + | // recorded returns the write count and the most recently written name/data, | |||
| 1138 | + | // captured under the store's own lock so it is safe to call from the test | |||
| 1139 | + | // goroutine after the write has happened-before via awaitSnapshotClean. | |||
| 1140 | + | func (s *gatedSnapshotStore) recorded() (int, string, []byte) { | |||
| 1141 | + | s.mu.Lock() | |||
| 1142 | + | defer s.mu.Unlock() | |||
| 1143 | + | return s.writes, s.name, s.data | |||
| 1144 | + | } | |||
| 1145 | + | ||||
| 1110 | type neverReturningSnapshotStore struct { | 1146 | type neverReturningSnapshotStore struct { | ||
| 1111 | entered chan struct{} | 1147 | entered chan struct{} | ||
| 1112 | once sync.Once | 1148 | once sync.Once | ||