Merge pull request #116 from bnema/fix/snapshot-kill-test-race
test(daemon): fix macOS CI hang in kill-session snapshot test
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 | ||