1// Package noticefile persists undeliverable daemon notices as newline-
2// delimited JSON so they survive a daemon restart and can be shown at the
3// next client attach.
4package noticefile
5
6import (
7 "bufio"
8 "encoding/json"
9 "errors"
10 "fmt"
11 "io"
12 "os"
13 "path/filepath"
14 "sync"
15 "syscall"
16
17 "github.com/bnema/vev/internal/domain"
18 "github.com/bnema/vev/internal/ports"
19 "github.com/bnema/vev/pkg/safedir"
20)
21
22const (
23 fileName = "pending-notices.jsonl"
24 inFlightFileName = "pending-notices.inflight.jsonl"
25 lockFileName = "pending-notices.lock"
26 claimLockFileName = "pending-notices.claim.lock"
27 maxNoticeRecordSize = 4 << 20
28 noticeReadBufferSize = 64 << 10
29)
30
31// Store persists notices as JSON-lines under dir.
32type Store struct {
33 dir string
34
35 claimMu sync.Mutex
36 claimLock *os.File
37
38 // afterAppendOpen is used by package tests to pause an append after it has
39 // opened the pending file while it owns the interprocess lock. It is nil in
40 // production.
41 afterAppendOpen func()
42
43 // removeFile, syncDirectory, and beforeAckRelease make Ack cleanup and
44 // claim-release ordering deterministic in package tests. They are nil in
45 // production.
46 removeFile func(string) error
47 syncDirectory func(string) error
48 beforeAckRelease func()
49}
50
51var (
52 // ErrClaimInProgress means another live Store owns the in-flight claim.
53 // Callers must leave it for that owner rather than replaying or acknowledging
54 // its notices.
55 ErrClaimInProgress = errors.New("pending notice claim in progress")
56 // ErrNoClaimOwner means Ack was called by a Store that did not make the
57 // current claim.
58 ErrNoClaimOwner = errors.New("pending notice claim is not owned by this store")
59)
60
61var _ ports.NoticeStore = (*Store)(nil)
62
63// New creates a file-backed notice store rooted at dir.
64func New(dir string) *Store {
65 return &Store{dir: dir}
66}
67
68func (s *Store) path() string {
69 return filepath.Join(s.dir, fileName)
70}
71
72func (s *Store) inFlightPath() string {
73 return filepath.Join(s.dir, inFlightFileName)
74}
75
76func (s *Store) lockPath() string {
77 return filepath.Join(s.dir, lockFileName)
78}
79
80func (s *Store) claimLockPath() string {
81 return filepath.Join(s.dir, claimLockFileName)
82}
83
84// withLock serializes Append, Claim, and Ack across Store instances and
85// processes. Holding this lock while Claim rotates and reads the file prevents
86// an appender with an already-open file descriptor from writing into a claim
87// after its reader reached EOF.
88func (s *Store) withLock(fn func() error) error {
89 if err := safedir.EnsurePrivate(s.dir); err != nil {
90 return fmt.Errorf("create notice dir: %w", err)
91 }
92 f, err := os.OpenFile(s.lockPath(), os.O_CREATE|os.O_RDWR, 0o600)
93 if err != nil {
94 return fmt.Errorf("open notice lock: %w", err)
95 }
96 defer func() { _ = f.Close() }()
97 if err := f.Chmod(0o600); err != nil {
98 return fmt.Errorf("secure notice lock: %w", err)
99 }
100 if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX); err != nil {
101 return fmt.Errorf("lock pending notices: %w", err)
102 }
103 defer func() { _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN) }()
104 return fn()
105}
106
107// acquireClaimLock reserves the current in-flight file for this Store until
108// Ack. The flock is released by the kernel if this process exits, allowing a
109// later daemon to recover an abandoned claim.
110func (s *Store) acquireClaimLock() error {
111 s.claimMu.Lock()
112 defer s.claimMu.Unlock()
113 if s.claimLock != nil {
114 return ErrClaimInProgress
115 }
116 f, err := os.OpenFile(s.claimLockPath(), os.O_CREATE|os.O_RDWR, 0o600)
117 if err != nil {
118 return fmt.Errorf("open notice claim lock: %w", err)
119 }
120 if err := f.Chmod(0o600); err != nil {
121 _ = f.Close()
122 return fmt.Errorf("secure notice claim lock: %w", err)
123 }
124 if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err != nil {
125 _ = f.Close()
126 if errors.Is(err, syscall.EWOULDBLOCK) {
127 return ErrClaimInProgress
128 }
129 return fmt.Errorf("lock pending notice claim: %w", err)
130 }
131 s.claimLock = f
132 return nil
133}
134
135func (s *Store) releaseClaimLock() {
136 s.claimMu.Lock()
137 f := s.claimLock
138 s.claimLock = nil
139 s.claimMu.Unlock()
140 if f == nil {
141 return
142 }
143 _ = syscall.Flock(int(f.Fd()), syscall.LOCK_UN)
144 _ = f.Close()
145}
146
147func (s *Store) ownsClaim() bool {
148 s.claimMu.Lock()
149 defer s.claimMu.Unlock()
150 return s.claimLock != nil
151}
152
153func syncDir(dir string) error {
154 f, err := os.Open(dir)
155 if err != nil {
156 return err
157 }
158 defer func() { _ = f.Close() }()
159 return f.Sync()
160}
161
162// Append writes n as one JSON line, creating the private state dir if needed.
163// A successful append is synced before it is made available to a later claim.
164func (s *Store) Append(n domain.Notification) error {
165 data, err := json.Marshal(n)
166 if err != nil {
167 return fmt.Errorf("marshal notice: %w", err)
168 }
169 data = append(data, '\n')
170
171 return s.withLock(func() error {
172 f, err := os.OpenFile(s.path(), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o600)
173 if err != nil {
174 return fmt.Errorf("open pending notices: %w", err)
175 }
176 if err := f.Chmod(0o600); err != nil {
177 _ = f.Close()
178 return fmt.Errorf("secure pending notices: %w", err)
179 }
180 if s.afterAppendOpen != nil {
181 s.afterAppendOpen()
182 }
183 if _, err := f.Write(data); err != nil {
184 _ = f.Close()
185 return fmt.Errorf("write notice: %w", err)
186 }
187 if err := f.Sync(); err != nil {
188 _ = f.Close()
189 return fmt.Errorf("sync pending notices: %w", err)
190 }
191 if err := f.Close(); err != nil {
192 return fmt.Errorf("close pending notices: %w", err)
193 }
194 if err := syncDir(s.dir); err != nil {
195 return fmt.Errorf("sync notice dir: %w", err)
196 }
197 return nil
198 })
199}
200
201// Claim atomically moves pending notices into an in-flight file and returns
202// their valid JSONL entries in order. Repeating Claim on its owning Store
203// replays that Store's existing claim; a concurrent Store cannot replay or
204// acknowledge it. The ownership flock is released on process exit, allowing
205// an unacknowledged import to be replayed after a crash. A missing pending
206// file means there is nothing to claim.
207func (s *Store) Claim() ([]domain.Notification, error) {
208 var notices []domain.Notification
209 err := s.withLock(func() error {
210 if s.ownsClaim() {
211 var err error
212 notices, err = readNotices(s.inFlightPath())
213 return err
214 }
215 if err := s.acquireClaimLock(); err != nil {
216 return err
217 }
218 keepClaim := false
219 defer func() {
220 if !keepClaim {
221 s.releaseClaimLock()
222 }
223 }()
224
225 inFlight := s.inFlightPath()
226 if _, err := os.Stat(inFlight); err != nil {
227 if !errors.Is(err, os.ErrNotExist) {
228 return fmt.Errorf("stat in-flight notices: %w", err)
229 }
230 if err := os.Rename(s.path(), inFlight); err != nil {
231 if errors.Is(err, os.ErrNotExist) {
232 return nil
233 }
234 return fmt.Errorf("claim pending notices: %w", err)
235 }
236 // fsync the parent directory so the atomic claim is recoverable after
237 // power loss, not merely after a clean process restart.
238 if err := syncDir(s.dir); err != nil {
239 return fmt.Errorf("sync claimed notices: %w", err)
240 }
241 }
242
243 var err error
244 notices, err = readNotices(inFlight)
245 if err == nil {
246 keepClaim = true
247 }
248 return err
249 })
250 if err != nil {
251 return nil, err
252 }
253 return notices, nil
254}
255
256func readNotices(path string) ([]domain.Notification, error) {
257 f, err := os.Open(path)
258 if err != nil {
259 return nil, fmt.Errorf("open claimed notices: %w", err)
260 }
261 defer func() { _ = f.Close() }()
262
263 var notices []domain.Notification
264 reader := bufio.NewReaderSize(f, noticeReadBufferSize)
265 for {
266 line, oversized, eof, err := readNoticeRecord(reader)
267 if err != nil {
268 return nil, fmt.Errorf("read claimed notices: %w", err)
269 }
270 if len(line) > 0 && !oversized {
271 var n domain.Notification
272 if err := json.Unmarshal(line, &n); err == nil {
273 notices = append(notices, n)
274 }
275 }
276 if eof {
277 return notices, nil
278 }
279 }
280}
281
282// readNoticeRecord reads one newline-delimited record without retaining more
283// than maxNoticeRecordSize bytes. Oversized records are discarded through
284// their delimiter so a later valid record can still be read.
285func readNoticeRecord(reader *bufio.Reader) (line []byte, oversized, eof bool, err error) {
286 for {
287 fragment, readErr := reader.ReadSlice('\n')
288 if len(fragment) > 0 && fragment[len(fragment)-1] == '\n' {
289 fragment = fragment[:len(fragment)-1]
290 }
291 if !oversized {
292 line, oversized = appendNoticeFragment(line, fragment)
293 }
294
295 switch {
296 case readErr == nil:
297 return line, oversized, false, nil
298 case errors.Is(readErr, bufio.ErrBufferFull):
299 continue
300 case errors.Is(readErr, io.EOF):
301 return line, oversized, true, nil
302 default:
303 return nil, false, false, readErr
304 }
305 }
306}
307
308func appendNoticeFragment(line, fragment []byte) ([]byte, bool) {
309 if len(fragment) > maxNoticeRecordSize-len(line) {
310 return nil, true
311 }
312 required := len(line) + len(fragment)
313 if required > cap(line) {
314 capacity := max(cap(line)*2, noticeReadBufferSize)
315 capacity = min(capacity, maxNoticeRecordSize)
316 grown := make([]byte, len(line), capacity)
317 copy(grown, line)
318 line = grown
319 }
320 return append(line, fragment...), false
321}
322
323// Ack permanently removes this Store's in-flight claim. It is only called once
324// the daemon has recorded and queued every claimed notice.
325func (s *Store) Ack() error {
326 if !s.ownsClaim() {
327 return ErrNoClaimOwner
328 }
329 return s.withLock(func() error {
330 if !s.ownsClaim() {
331 return ErrNoClaimOwner
332 }
333 // Ack is terminal for this Store's claim even when removing or syncing
334 // the record fails. Release while the operation lock is still held so a
335 // concurrent Claim on this Store cannot see a removed in-flight file
336 // while it still appears to own the claim.
337 defer func() {
338 if s.beforeAckRelease != nil {
339 s.beforeAckRelease()
340 }
341 s.releaseClaimLock()
342 }()
343
344 remove := os.Remove
345 if s.removeFile != nil {
346 remove = s.removeFile
347 }
348 if err := remove(s.inFlightPath()); err != nil && !errors.Is(err, os.ErrNotExist) {
349 return fmt.Errorf("ack claimed notices: %w", err)
350 }
351 sync := syncDir
352 if s.syncDirectory != nil {
353 sync = s.syncDirectory
354 }
355 if err := sync(s.dir); err != nil {
356 return fmt.Errorf("sync acknowledged notices: %w", err)
357 }
358 return nil
359 })
360}