fix(backend): report ingest correctness: 500 on a refused report, 413 on oversize, global body cap (closes #23)
check / check (push) Successful in 10s
check / check (push) Successful in 10s
A report the buffer refuses now returns 500 instead of a false `ok`. Reports reach disk later, so a failed disk write is still answered 200 and shows in the log, and at shutdown as a failed stop with a non-zero exit. An over-limit body returns 413; malformed JSON stays 400. A MaxBodyBytes middleware caps every route at 1 MiB; a route group can only lower that limit. The raw geo blob is no longer logged; client_id, timestamp and decode error text are cut to 128 bytes before logging. Panic recovery logs the panic value and stack through slog. Model: opus-5-5
This commit was merged in pull request #58.
This commit is contained in:
@@ -80,12 +80,16 @@ func New(
|
||||
// stopOnce makes OnStop idempotent: a second
|
||||
// invocation must not close an already-closed channel
|
||||
// (which would panic) or flush again.
|
||||
var err error
|
||||
|
||||
b.stopOnce.Do(func() {
|
||||
close(b.done)
|
||||
b.flushLocked()
|
||||
err = b.flushLocked()
|
||||
})
|
||||
|
||||
return nil
|
||||
// A failed final flush fails the stop, so the process
|
||||
// exits non-zero.
|
||||
return err
|
||||
},
|
||||
})
|
||||
|
||||
@@ -109,7 +113,12 @@ func (b *Buffer) Append(v any) error {
|
||||
data := b.drainBuf()
|
||||
b.mu.Unlock()
|
||||
|
||||
go b.writeFile(data)
|
||||
go func() {
|
||||
writeErr := b.writeFile(data)
|
||||
if writeErr != nil {
|
||||
b.log.Error("flush reports failed", "error", writeErr)
|
||||
}
|
||||
}()
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -128,7 +137,10 @@ func (b *Buffer) flushLoop() {
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
b.flushLocked()
|
||||
err := b.flushLocked()
|
||||
if err != nil {
|
||||
b.log.Error("flush reports failed", "error", err)
|
||||
}
|
||||
case <-b.done:
|
||||
return
|
||||
}
|
||||
@@ -137,19 +149,19 @@ func (b *Buffer) flushLoop() {
|
||||
|
||||
// flushLocked acquires the lock, drains the buffer, and
|
||||
// writes the data to a compressed file.
|
||||
func (b *Buffer) flushLocked() {
|
||||
func (b *Buffer) flushLocked() error {
|
||||
b.mu.Lock()
|
||||
|
||||
if b.buf.Len() == 0 {
|
||||
b.mu.Unlock()
|
||||
|
||||
return
|
||||
return nil
|
||||
}
|
||||
|
||||
data := b.drainBuf()
|
||||
b.mu.Unlock()
|
||||
|
||||
b.writeFile(data)
|
||||
return b.writeFile(data)
|
||||
}
|
||||
|
||||
// drainBuf copies the buffer contents and resets it.
|
||||
@@ -164,7 +176,7 @@ func (b *Buffer) drainBuf() []byte {
|
||||
|
||||
// writeFile creates a timestamped zstd-compressed JSONL file
|
||||
// in the data directory.
|
||||
func (b *Buffer) writeFile(data []byte) {
|
||||
func (b *Buffer) writeFile(data []byte) error {
|
||||
ts := time.Now().UTC().Format("2006-01-02T15-04-05.000Z")
|
||||
name := fmt.Sprintf("reports-%s.jsonl.zst", ts)
|
||||
path := filepath.Join(b.dataDir, name)
|
||||
@@ -177,31 +189,35 @@ func (b *Buffer) writeFile(data []byte) {
|
||||
filePerms,
|
||||
)
|
||||
if err != nil {
|
||||
b.log.Error("create report file", "error", err)
|
||||
|
||||
return
|
||||
return fmt.Errorf("create report file: %w", err)
|
||||
}
|
||||
|
||||
// Closes the file on the early returns below. The success
|
||||
// path closes it explicitly to check the error; closing it
|
||||
// a second time here is harmless.
|
||||
defer func() { _ = f.Close() }()
|
||||
|
||||
enc, err := zstd.NewWriter(f)
|
||||
if err != nil {
|
||||
b.log.Error("create zstd encoder", "error", err)
|
||||
|
||||
return
|
||||
return fmt.Errorf("create zstd encoder: %w", err)
|
||||
}
|
||||
|
||||
_, writeErr := enc.Write(data)
|
||||
if writeErr != nil {
|
||||
b.log.Error("write compressed data", "error", writeErr)
|
||||
|
||||
_, err = enc.Write(data)
|
||||
if err != nil {
|
||||
_ = enc.Close()
|
||||
|
||||
return
|
||||
return fmt.Errorf("write compressed data: %w", err)
|
||||
}
|
||||
|
||||
closeErr := enc.Close()
|
||||
if closeErr != nil {
|
||||
b.log.Error("close zstd encoder", "error", closeErr)
|
||||
err = enc.Close()
|
||||
if err != nil {
|
||||
return fmt.Errorf("close zstd encoder: %w", err)
|
||||
}
|
||||
|
||||
err = f.Close()
|
||||
if err != nil {
|
||||
return fmt.Errorf("close report file: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
package reportbuf_test
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"io/fs"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -52,6 +54,46 @@ func TestFlushOnShutdown(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestFailedFinalFlushFailsStop proves a final flush that cannot
|
||||
// write its file makes the stop fail, which makes the process
|
||||
// exit non-zero instead of dropping the buffered reports silently.
|
||||
func TestFailedFinalFlushFailsStop(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
t.Setenv("DATA_DIR", dir)
|
||||
|
||||
var buf *reportbuf.Buffer
|
||||
|
||||
app := fxtest.New(t,
|
||||
fx.Provide(
|
||||
globals.New,
|
||||
logger.New,
|
||||
config.New,
|
||||
reportbuf.New,
|
||||
),
|
||||
fx.Populate(&buf),
|
||||
)
|
||||
|
||||
app.RequireStart()
|
||||
|
||||
err := buf.Append(map[string]string{"probe": "shutdown"})
|
||||
if err != nil {
|
||||
t.Fatalf("append report: %v", err)
|
||||
}
|
||||
|
||||
// Removing the data directory leaves the final flush nowhere to
|
||||
// write. A read-only directory would not do: tests run as root
|
||||
// in the backend image, and root ignores the read-only bit.
|
||||
err = os.RemoveAll(dir)
|
||||
if err != nil {
|
||||
t.Fatalf("remove data dir: %v", err)
|
||||
}
|
||||
|
||||
err = app.Stop(t.Context())
|
||||
if !errors.Is(err, fs.ErrNotExist) {
|
||||
t.Fatalf("stop error = %v, want the final flush's error", err)
|
||||
}
|
||||
}
|
||||
|
||||
func hasReportFile(t *testing.T, dir string) bool {
|
||||
t.Helper()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user