detached 156e0f0

fix(noticefile): recover after oversized pending records

commit
156e0f0f771b40e421a02eb47697a45e912239e9
author
brice · · 2026-07-21 11:26 UTC
parent
fa73472
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
2 files · +119 -16
Unified
UnifiedSplit
M internal/adapters/noticefile/noticefile.go +64 -16
@@ -8,6 +8,7 @@ import (
88 "encoding/json"
99 "errors"
1010 "fmt"
11+ "io"
1112 "os"
1213 "path/filepath"
1314 "syscall"
@@ -18,9 +19,11 @@ import (
1819 )
1920
2021 const (
21- fileName = "pending-notices.jsonl"
22- inFlightFileName = "pending-notices.inflight.jsonl"
23- lockFileName = "pending-notices.lock"
22+ fileName = "pending-notices.jsonl"
23+ inFlightFileName = "pending-notices.inflight.jsonl"
24+ lockFileName = "pending-notices.lock"
25+ maxNoticeRecordSize = 4 << 20
26+ noticeReadBufferSize = 64 << 10
2427 )
2528
2629 // Store persists notices as JSON-lines under dir.
@@ -166,23 +169,68 @@ func readNotices(path string) ([]domain.Notification, error) {
166169 defer func() { _ = f.Close() }()
167170
168171 var notices []domain.Notification
169- scanner := bufio.NewScanner(f)
170- scanner.Buffer(make([]byte, 0, 64*1024), 4<<20)
171- for scanner.Scan() {
172- line := scanner.Bytes()
173- if len(line) == 0 {
174- continue
172+ reader := bufio.NewReaderSize(f, noticeReadBufferSize)
173+ for {
174+ line, oversized, eof, err := readNoticeRecord(reader)
175+ if err != nil {
176+ return nil, fmt.Errorf("read claimed notices: %w", err)
177+ }
178+ if len(line) > 0 && !oversized {
179+ var n domain.Notification
180+ if err := json.Unmarshal(line, &n); err == nil {
181+ notices = append(notices, n)
182+ }
183+ }
184+ if eof {
185+ return notices, nil
186+ }
187+ }
188+}
189+
190+// readNoticeRecord reads one newline-delimited record without retaining more
191+// than maxNoticeRecordSize bytes. Oversized records are discarded through
192+// their delimiter so a later valid record can still be read.
193+func readNoticeRecord(reader *bufio.Reader) (line []byte, oversized, eof bool, err error) {
194+ for {
195+ fragment, readErr := reader.ReadSlice('\n')
196+ if len(fragment) > 0 && fragment[len(fragment)-1] == '\n' {
197+ fragment = fragment[:len(fragment)-1]
175198 }
176- var n domain.Notification
177- if err := json.Unmarshal(line, &n); err != nil {
178- continue // corrupt or partial line must not abort the claim
199+ if !oversized {
200+ line, oversized = appendNoticeFragment(line, fragment)
201+ }
202+
203+ switch {
204+ case readErr == nil:
205+ return line, oversized, false, nil
206+ case errors.Is(readErr, bufio.ErrBufferFull):
207+ continue
208+ case errors.Is(readErr, io.EOF):
209+ return line, oversized, true, nil
210+ default:
211+ return nil, false, false, readErr
179212 }
180- notices = append(notices, n)
181213 }
182- if err := scanner.Err(); err != nil {
183- return nil, fmt.Errorf("read claimed notices: %w", err)
214+}
215+
216+func appendNoticeFragment(line, fragment []byte) ([]byte, bool) {
217+ if len(fragment) > maxNoticeRecordSize-len(line) {
218+ return nil, true
184219 }
185- return notices, nil
220+ required := len(line) + len(fragment)
221+ if required > cap(line) {
222+ capacity := cap(line) * 2
223+ if capacity < noticeReadBufferSize {
224+ capacity = noticeReadBufferSize
225+ }
226+ if capacity > maxNoticeRecordSize {
227+ capacity = maxNoticeRecordSize
228+ }
229+ grown := make([]byte, len(line), capacity)
230+ copy(grown, line)
231+ line = grown
232+ }
233+ return append(line, fragment...), false
186234 }
187235
188236 // Ack permanently removes the current in-flight claim. It is only called once
@@ -8,6 +8,7 @@ import (
8 "encoding/json"8 "encoding/json"
9 "errors"9 "errors"
10 "fmt"10 "fmt"
11+ "io"
11 "os"12 "os"
12 "path/filepath"13 "path/filepath"
13 "syscall"14 "syscall"
@@ -18,9 +19,11 @@ import (
18 )19 )
19 20
20 const (21 const (
21- fileName = "pending-notices.jsonl"22+ fileName = "pending-notices.jsonl"
22- inFlightFileName = "pending-notices.inflight.jsonl"23+ inFlightFileName = "pending-notices.inflight.jsonl"
23- lockFileName = "pending-notices.lock"24+ lockFileName = "pending-notices.lock"
25+ maxNoticeRecordSize = 4 << 20
26+ noticeReadBufferSize = 64 << 10
24 )27 )
25 28
26 // Store persists notices as JSON-lines under dir.29 // Store persists notices as JSON-lines under dir.
@@ -166,23 +169,68 @@ func readNotices(path string) ([]domain.Notification, error) {
166 defer func() { _ = f.Close() }()169 defer func() { _ = f.Close() }()
167 170
168 var notices []domain.Notification171 var notices []domain.Notification
169- scanner := bufio.NewScanner(f)172+ reader := bufio.NewReaderSize(f, noticeReadBufferSize)
170- scanner.Buffer(make([]byte, 0, 64*1024), 4<<20)173+ for {
171- for scanner.Scan() {174+ line, oversized, eof, err := readNoticeRecord(reader)
172- line := scanner.Bytes()175+ if err != nil {
173- if len(line) == 0 {176+ return nil, fmt.Errorf("read claimed notices: %w", err)
174- continue177+ }
178+ if len(line) > 0 && !oversized {
179+ var n domain.Notification
180+ if err := json.Unmarshal(line, &n); err == nil {
181+ notices = append(notices, n)
182+ }
183+ }
184+ if eof {
185+ return notices, nil
186+ }
187+ }
188+}
189+
190+// readNoticeRecord reads one newline-delimited record without retaining more
191+// than maxNoticeRecordSize bytes. Oversized records are discarded through
192+// their delimiter so a later valid record can still be read.
193+func readNoticeRecord(reader *bufio.Reader) (line []byte, oversized, eof bool, err error) {
194+ for {
195+ fragment, readErr := reader.ReadSlice('\n')
196+ if len(fragment) > 0 && fragment[len(fragment)-1] == '\n' {
197+ fragment = fragment[:len(fragment)-1]
175 }198 }
176- var n domain.Notification199+ if !oversized {
177- if err := json.Unmarshal(line, &n); err != nil {200+ line, oversized = appendNoticeFragment(line, fragment)
178- continue // corrupt or partial line must not abort the claim201+ }
202+
203+ switch {
204+ case readErr == nil:
205+ return line, oversized, false, nil
206+ case errors.Is(readErr, bufio.ErrBufferFull):
207+ continue
208+ case errors.Is(readErr, io.EOF):
209+ return line, oversized, true, nil
210+ default:
211+ return nil, false, false, readErr
179 }212 }
180- notices = append(notices, n)
181 }213 }
182- if err := scanner.Err(); err != nil {214+}
183- return nil, fmt.Errorf("read claimed notices: %w", err)215+
216+func appendNoticeFragment(line, fragment []byte) ([]byte, bool) {
217+ if len(fragment) > maxNoticeRecordSize-len(line) {
218+ return nil, true
184 }219 }
185- return notices, nil220+ required := len(line) + len(fragment)
221+ if required > cap(line) {
222+ capacity := cap(line) * 2
223+ if capacity < noticeReadBufferSize {
224+ capacity = noticeReadBufferSize
225+ }
226+ if capacity > maxNoticeRecordSize {
227+ capacity = maxNoticeRecordSize
228+ }
229+ grown := make([]byte, len(line), capacity)
230+ copy(grown, line)
231+ line = grown
232+ }
233+ return append(line, fragment...), false
186 }234 }
187 235
188 // Ack permanently removes the current in-flight claim. It is only called once236 // Ack permanently removes the current in-flight claim. It is only called once
M internal/adapters/noticefile/noticefile_test.go +55 -0
@@ -1,8 +1,10 @@
11 package noticefile
22
33 import (
4+ "encoding/json"
45 "os"
56 "path/filepath"
7+ "strings"
68 "testing"
79 "time"
810
@@ -116,6 +118,59 @@ func TestStoreClaimSkipsGarbageLinesInOrder(t *testing.T) {
116118 require.NoError(t, store.Ack())
117119 }
118120
121+func TestStoreClaimSkipsOversizedLines(t *testing.T) {
122+ t.Parallel()
123+
124+ validLine := func(message string) string {
125+ t.Helper()
126+ data, err := json.Marshal(domain.Notification{Message: message})
127+ require.NoError(t, err)
128+ return string(data)
129+ }
130+ const jsonMessagePrefix = `{"Code":0,"Severity":0,"Message":"`
131+ const jsonMessageSuffix = `","Details":"","Time":"0001-01-01T00:00:00Z","Count":0,"SessionID":""}`
132+ boundaryMessage := strings.Repeat("x", maxNoticeRecordSize-len(jsonMessagePrefix)-len(jsonMessageSuffix))
133+
134+ for _, tt := range []struct {
135+ name string
136+ lines []string
137+ want []string
138+ }{
139+ {
140+ name: "recovers after oversized line",
141+ lines: []string{validLine("first"), validLine(strings.Repeat("x", len(boundaryMessage)+1)), validLine("second")},
142+ want: []string{"first", "second"},
143+ },
144+ {
145+ name: "skips oversized final unterminated line",
146+ lines: []string{validLine("first"), validLine(strings.Repeat("x", len(boundaryMessage)+1))},
147+ want: []string{"first"},
148+ },
149+ {
150+ name: "accepts record at size limit",
151+ lines: []string{validLine(boundaryMessage)},
152+ want: []string{boundaryMessage},
153+ },
154+ } {
155+ t.Run(tt.name, func(t *testing.T) {
156+ dir := filepath.Join(t.TempDir(), "vev")
157+ records := strings.Join(tt.lines, "\n")
158+ if tt.name != "skips oversized final unterminated line" {
159+ records += "\n"
160+ }
161+ require.NoError(t, os.MkdirAll(dir, 0o700))
162+ require.NoError(t, os.WriteFile(filepath.Join(dir, fileName), []byte(records), 0o600))
163+
164+ got, err := New(dir).Claim()
165+ require.NoError(t, err)
166+ require.Len(t, got, len(tt.want))
167+ for i, want := range tt.want {
168+ require.Equal(t, want, got[i].Message)
169+ }
170+ })
171+ }
172+}
173+
119174 func TestStoreAppendUsesPrivatePermissions(t *testing.T) {
120175 t.Parallel()
121176 dir := filepath.Join(t.TempDir(), "vev")
@@ -1,8 +1,10 @@
1 package noticefile1 package noticefile
2 2
3 import (3 import (
4+ "encoding/json"
4 "os"5 "os"
5 "path/filepath"6 "path/filepath"
7+ "strings"
6 "testing"8 "testing"
7 "time"9 "time"
8 10
@@ -116,6 +118,59 @@ func TestStoreClaimSkipsGarbageLinesInOrder(t *testing.T) {
116 require.NoError(t, store.Ack())118 require.NoError(t, store.Ack())
117 }119 }
118 120
121+func TestStoreClaimSkipsOversizedLines(t *testing.T) {
122+ t.Parallel()
123+
124+ validLine := func(message string) string {
125+ t.Helper()
126+ data, err := json.Marshal(domain.Notification{Message: message})
127+ require.NoError(t, err)
128+ return string(data)
129+ }
130+ const jsonMessagePrefix = `{"Code":0,"Severity":0,"Message":"`
131+ const jsonMessageSuffix = `","Details":"","Time":"0001-01-01T00:00:00Z","Count":0,"SessionID":""}`
132+ boundaryMessage := strings.Repeat("x", maxNoticeRecordSize-len(jsonMessagePrefix)-len(jsonMessageSuffix))
133+
134+ for _, tt := range []struct {
135+ name string
136+ lines []string
137+ want []string
138+ }{
139+ {
140+ name: "recovers after oversized line",
141+ lines: []string{validLine("first"), validLine(strings.Repeat("x", len(boundaryMessage)+1)), validLine("second")},
142+ want: []string{"first", "second"},
143+ },
144+ {
145+ name: "skips oversized final unterminated line",
146+ lines: []string{validLine("first"), validLine(strings.Repeat("x", len(boundaryMessage)+1))},
147+ want: []string{"first"},
148+ },
149+ {
150+ name: "accepts record at size limit",
151+ lines: []string{validLine(boundaryMessage)},
152+ want: []string{boundaryMessage},
153+ },
154+ } {
155+ t.Run(tt.name, func(t *testing.T) {
156+ dir := filepath.Join(t.TempDir(), "vev")
157+ records := strings.Join(tt.lines, "\n")
158+ if tt.name != "skips oversized final unterminated line" {
159+ records += "\n"
160+ }
161+ require.NoError(t, os.MkdirAll(dir, 0o700))
162+ require.NoError(t, os.WriteFile(filepath.Join(dir, fileName), []byte(records), 0o600))
163+
164+ got, err := New(dir).Claim()
165+ require.NoError(t, err)
166+ require.Len(t, got, len(tt.want))
167+ for i, want := range tt.want {
168+ require.Equal(t, want, got[i].Message)
169+ }
170+ })
171+ }
172+}
173+
119 func TestStoreAppendUsesPrivatePermissions(t *testing.T) {174 func TestStoreAppendUsesPrivatePermissions(t *testing.T) {
120 t.Parallel()175 t.Parallel()
121 dir := filepath.Join(t.TempDir(), "vev")176 dir := filepath.Join(t.TempDir(), "vev")

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close