1package noticefile
2
3import (
4 "encoding/json"
5 "errors"
6 "os"
7 "path/filepath"
8 "strings"
9 "syscall"
10 "testing"
11 "time"
12
13 "github.com/stretchr/testify/require"
14
15 "github.com/bnema/vev/internal/domain"
16)
17
18func TestStoreClaimAckRoundTrip(t *testing.T) {
19 t.Parallel()
20 dir := filepath.Join(t.TempDir(), "vev")
21 store := New(dir)
22
23 first := domain.Notification{
24 Code: domain.NoticeSnapshotWrite,
25 Severity: domain.NoticeError,
26 Message: "session foo shut down without saving terminal state",
27 Details: "boom: disk full",
28 Time: time.Date(2026, 7, 20, 10, 0, 0, 0, time.UTC),
29 Count: 1,
30 SessionID: "sess-1",
31 }
32 second := domain.Notification{
33 Code: domain.NoticeSnapshotRestore,
34 Severity: domain.NoticeWarn,
35 Message: "session bar could not be restored",
36 Time: time.Date(2026, 7, 20, 10, 5, 0, 0, time.UTC),
37 Count: 2,
38 }
39 require.NoError(t, store.Append(first))
40 require.NoError(t, store.Append(second))
41
42 claimed, err := store.Claim()
43 require.NoError(t, err)
44 require.Equal(t, []domain.Notification{first, second}, claimed)
45
46 // Releasing the claim lock simulates the kernel closing it when the daemon
47 // crashes. The in-flight file remains and a fresh process/store replays it.
48 store.releaseClaimLock()
49 recovered := New(dir)
50 replayed, err := recovered.Claim()
51 require.NoError(t, err)
52 require.Equal(t, claimed, replayed)
53
54 require.NoError(t, recovered.Ack())
55 again, err := New(dir).Claim()
56 require.NoError(t, err)
57 require.Empty(t, again)
58}
59
60func TestStoreClaimWaitsForAppendInCriticalSection(t *testing.T) {
61 t.Parallel()
62 dir := filepath.Join(t.TempDir(), "vev")
63 store := New(dir)
64 first := domain.Notification{Message: "before claim"}
65 second := domain.Notification{Message: "append holds pending file open"}
66 require.NoError(t, store.Append(first))
67
68 appendEntered := make(chan struct{})
69 allowAppend := make(chan struct{})
70 store.afterAppendOpen = func() {
71 close(appendEntered)
72 <-allowAppend
73 }
74 appendDone := make(chan error, 1)
75 go func() { appendDone <- store.Append(second) }()
76 <-appendEntered // Append owns the operation lock and an open pending file.
77
78 claimant := New(dir)
79 claimDone := make(chan []domain.Notification, 1)
80 errDone := make(chan error, 1)
81 go func() {
82 claimed, err := claimant.Claim()
83 claimDone <- claimed
84 errDone <- err
85 }()
86 close(allowAppend)
87 require.NoError(t, <-appendDone)
88 require.NoError(t, <-errDone)
89 require.Equal(t, []domain.Notification{first, second}, <-claimDone)
90 require.NoError(t, claimant.Ack())
91}
92
93func TestStoreRepeatedClaimReplaysOwnerAndBlocksCompetitor(t *testing.T) {
94 t.Parallel()
95 dir := filepath.Join(t.TempDir(), "vev")
96 owner := New(dir)
97 want := domain.Notification{Message: "owned"}
98 require.NoError(t, owner.Append(want))
99
100 first, err := owner.Claim()
101 require.NoError(t, err)
102 require.Equal(t, []domain.Notification{want}, first)
103
104 replayed, err := owner.Claim()
105 require.NoError(t, err)
106 require.Equal(t, first, replayed)
107
108 other := New(dir)
109 blocked, err := other.Claim()
110 require.ErrorIs(t, err, ErrClaimInProgress)
111 require.Nil(t, blocked)
112
113 require.NoError(t, owner.Ack())
114 empty, err := other.Claim()
115 require.NoError(t, err)
116 require.Empty(t, empty)
117}
118
119func TestStoreAckReleasesClaimAfterCleanupFailure(t *testing.T) {
120 t.Parallel()
121
122 for _, tt := range []struct {
123 name string
124 inject func(*Store, error)
125 want []domain.Notification
126 }{
127 {
128 name: "remove",
129 inject: func(store *Store, cause error) {
130 store.removeFile = func(string) error { return cause }
131 },
132 want: []domain.Notification{{Message: "owned"}},
133 },
134 {
135 name: "sync directory",
136 inject: func(store *Store, cause error) {
137 store.syncDirectory = func(string) error { return cause }
138 },
139 want: nil,
140 },
141 } {
142 t.Run(tt.name, func(t *testing.T) {
143 dir := filepath.Join(t.TempDir(), "vev")
144 owner := New(dir)
145 require.NoError(t, owner.Append(domain.Notification{Message: "owned"}))
146 _, err := owner.Claim()
147 require.NoError(t, err)
148
149 cause := errors.New("cleanup failed")
150 tt.inject(owner, cause)
151 require.ErrorIs(t, owner.Ack(), cause)
152
153 // A new Store proves the failed Ack released the ownership flock.
154 recovered, err := New(dir).Claim()
155 require.NoError(t, err)
156 require.Equal(t, tt.want, recovered)
157 })
158 }
159}
160
161func TestStoreAckSerializesClaimReleaseWithSameStoreReclaim(t *testing.T) {
162 dir := filepath.Join(t.TempDir(), "vev")
163 store := New(dir)
164 require.NoError(t, store.Append(domain.Notification{Message: "owned"}))
165 _, err := store.Claim()
166 require.NoError(t, err)
167
168 releaseEntered := make(chan struct{})
169 allowRelease := make(chan struct{})
170 lockHeld := make(chan error, 1)
171 store.beforeAckRelease = func() {
172 held, err := operationLockHeld(store.lockPath())
173 if err != nil {
174 lockHeld <- err
175 } else if !held {
176 lockHeld <- errors.New("Ack released the operation lock before claim ownership")
177 } else {
178 lockHeld <- nil
179 }
180 close(releaseEntered)
181 <-allowRelease
182 }
183
184 ackDone := make(chan error, 1)
185 go func() { ackDone <- store.Ack() }()
186 <-releaseEntered // The in-flight file is gone, but Ack still owns the operation lock.
187 require.NoError(t, <-lockHeld)
188
189 claimStarted := make(chan struct{})
190 claimDone := make(chan struct {
191 notices []domain.Notification
192 err error
193 }, 1)
194 go func() {
195 close(claimStarted)
196 notices, err := store.Claim()
197 claimDone <- struct {
198 notices []domain.Notification
199 err error
200 }{notices, err}
201 }()
202 <-claimStarted
203 close(allowRelease)
204
205 require.NoError(t, <-ackDone)
206 reclaimed := <-claimDone
207 require.NoError(t, reclaimed.err)
208 require.Empty(t, reclaimed.notices)
209}
210
211func operationLockHeld(path string) (bool, error) {
212 f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600)
213 if err != nil {
214 return false, err
215 }
216 defer func() { _ = f.Close() }()
217 if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
218 if errors.Is(err, syscall.EWOULDBLOCK) {
219 return true, nil
220 }
221 return false, err
222 }
223 defer func() { _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) }()
224 return false, nil
225}
226
227func TestStoreClaimOwnershipPreventsLiveReplayOrAck(t *testing.T) {
228 t.Parallel()
229 dir := filepath.Join(t.TempDir(), "vev")
230 owner := New(dir)
231 require.NoError(t, owner.Append(domain.Notification{Message: "owned"}))
232 claimed, err := owner.Claim()
233 require.NoError(t, err)
234 require.Len(t, claimed, 1)
235
236 other := New(dir)
237 replayed, err := other.Claim()
238 require.ErrorIs(t, err, ErrClaimInProgress)
239 require.Nil(t, replayed)
240 require.ErrorIs(t, other.Ack(), ErrNoClaimOwner)
241
242 // The rejected Ack must leave the owner's in-flight record intact.
243 require.NoError(t, owner.Ack())
244 empty, err := New(dir).Claim()
245 require.NoError(t, err)
246 require.Empty(t, empty)
247}
248
249func TestStoreClaimMissingFileReturnsNil(t *testing.T) {
250 t.Parallel()
251
252 got, err := New(filepath.Join(t.TempDir(), "vev")).Claim()
253 require.NoError(t, err)
254 require.Nil(t, got)
255}
256
257func TestStoreClaimSkipsGarbageLinesInOrder(t *testing.T) {
258 t.Parallel()
259 dir := filepath.Join(t.TempDir(), "vev")
260 store := New(dir)
261 path := filepath.Join(dir, fileName)
262 require.NoError(t, os.MkdirAll(dir, 0o700))
263
264 good := `{"Code":9,"Severity":2,"Message":"first"}` + "\n"
265 garbage := "not json at all\n"
266 good2 := `{"Code":10,"Severity":1,"Message":"second"}` + "\n"
267 require.NoError(t, os.WriteFile(path, []byte(good+garbage+good2), 0o600))
268
269 got, err := store.Claim()
270 require.NoError(t, err)
271 require.Len(t, got, 2)
272 require.Equal(t, "first", got[0].Message)
273 require.Equal(t, "second", got[1].Message)
274 require.NoError(t, store.Ack())
275}
276
277func TestStoreClaimSkipsOversizedLines(t *testing.T) {
278 t.Parallel()
279
280 validLine := func(message string) string {
281 t.Helper()
282 data, err := json.Marshal(domain.Notification{Message: message})
283 require.NoError(t, err)
284 return string(data)
285 }
286 // Derive JSON overhead from the encoder rather than duplicating its current
287 // field layout here. The boundary payload uses ASCII so each added message
288 // byte remains one marshaled byte.
289 boundaryMessage := strings.Repeat("x", maxNoticeRecordSize-len(validLine("")))
290
291 for _, tt := range []struct {
292 name string
293 lines []string
294 want []string
295 }{
296 {
297 name: "recovers after oversized line",
298 lines: []string{validLine("first"), validLine(strings.Repeat("x", len(boundaryMessage)+1)), validLine("second")},
299 want: []string{"first", "second"},
300 },
301 {
302 name: "skips oversized final unterminated line",
303 lines: []string{validLine("first"), validLine(strings.Repeat("x", len(boundaryMessage)+1))},
304 want: []string{"first"},
305 },
306 {
307 name: "accepts record at size limit",
308 lines: []string{validLine(boundaryMessage)},
309 want: []string{boundaryMessage},
310 },
311 } {
312 t.Run(tt.name, func(t *testing.T) {
313 dir := filepath.Join(t.TempDir(), "vev")
314 records := strings.Join(tt.lines, "\n")
315 if tt.name != "skips oversized final unterminated line" {
316 records += "\n"
317 }
318 require.NoError(t, os.MkdirAll(dir, 0o700))
319 require.NoError(t, os.WriteFile(filepath.Join(dir, fileName), []byte(records), 0o600))
320
321 got, err := New(dir).Claim()
322 require.NoError(t, err)
323 require.Len(t, got, len(tt.want))
324 for i, want := range tt.want {
325 require.Equal(t, want, got[i].Message)
326 }
327 })
328 }
329}
330
331func TestStoreAppendUsesPrivatePermissions(t *testing.T) {
332 t.Parallel()
333 dir := filepath.Join(t.TempDir(), "vev")
334 store := New(dir)
335
336 require.NoError(t, store.Append(domain.Notification{Message: "x"}))
337 info, err := os.Stat(dir)
338 require.NoError(t, err)
339 require.Equal(t, os.FileMode(0o700), info.Mode().Perm())
340 fi, err := os.Stat(filepath.Join(dir, fileName))
341 require.NoError(t, err)
342 require.Equal(t, os.FileMode(0o600), fi.Mode().Perm())
343
344 _, err = store.Claim()
345 require.NoError(t, err)
346 fi, err = os.Stat(filepath.Join(dir, inFlightFileName))
347 require.NoError(t, err)
348 require.Equal(t, os.FileMode(0o600), fi.Mode().Perm())
349}