Delete the oldest report files to stay under the size cap (closes #54)
check / check (push) Successful in 1m28s

When a report would take the report files past DATA_DIR_MAX_BYTES,
reportbuf now deletes the oldest report files until it fits, and does
the same at start when files left by an earlier run are already past
it. A file joins the files that may be deleted only once it is
completely written, so a file still being written is never deleted. A
report is refused with 507 only when the reports waiting to be written
fill the cap on their own, and then no file is deleted. The reports of
a failed write stop counting, and the part of its file written is
removed. A file whose deletion fails keeps counting; one already
deleted by hand counts as freed.

Model: opus-5-5
This commit is contained in:
2026-10-03 14:12:07 +00:00
parent 0ab6418e3d
commit a2dd439812
8 changed files with 523 additions and 105 deletions
+1 -1
View File
@@ -218,7 +218,7 @@ What the [upaas](https://git.eeqj.de/sneak/upaas) app for netwatch needs:
- `REPORTS_PER_MINUTE`, default `60`: reports each client address may send a - `REPORTS_PER_MINUTE`, default `60`: reports each client address may send a
minute minute
- `DATA_DIR_MAX_BYTES`, default `1073741824` (1 GiB): the most room the - `DATA_DIR_MAX_BYTES`, default `1073741824` (1 GiB): the most room the
report files may take report files may take; the oldest are deleted to stay under it
- `CORS_ALLOWED_ORIGINS`, default empty: other origins whose pages may call - `CORS_ALLOWED_ORIGINS`, default empty: other origins whose pages may call
the API the API
- `DEBUG`, default `false`: debug logging - `DEBUG`, default `false`: debug logging
+9
View File
@@ -23,6 +23,15 @@ latest run passes.
# Completed Steps # Completed Steps
- 2026-10-03: `DATA_DIR_MAX_BYTES` is now how much of the report files is kept
(issue #54): when a report would take them past it, the oldest report files
are deleted to make room, each deletion logged, and at start files already
past it are deleted the same way. A file still being written is never deleted.
A report is refused with 507 only when the reports waiting to be written fill
the cap on their own, and then no file is deleted. The reports of a failed
write stop counting, and the part of its file written is removed. A file that
cannot be deleted still counts until the next start; one already deleted by
hand counts as freed
- 2026-10-03: each target check times out after 80% of the refresh interval - 2026-10-03: each target check times out after 80% of the refresh interval
(issue #78), 24 seconds at 30 seconds, where it was capped at 3 seconds. A (issue #78), 24 seconds at 30 seconds, where it was capped at 3 seconds. A
round started early, after an interval change or when the recovery probe finds round started early, after an interval change or when the recovery probe finds
+12 -7
View File
@@ -80,7 +80,7 @@ Internal packages in `internal/` follow standard Go project layout:
| `BIND_ADDRESS` | empty | IP address to listen on; empty listens on every interface | | `BIND_ADDRESS` | empty | IP address to listen on; empty listens on every interface |
| `PORT` | `8080` | HTTP listen port | | `PORT` | `8080` | HTTP listen port |
| `DATA_DIR` | `./data/reports` | Directory for compressed reports | | `DATA_DIR` | `./data/reports` | Directory for compressed reports |
| `DATA_DIR_MAX_BYTES` | `1073741824` (1 GiB) | Largest total size of the report files in `DATA_DIR`; see [Report limits](#report-limits) | | `DATA_DIR_MAX_BYTES` | `1073741824` (1 GiB) | Most bytes of report files kept in `DATA_DIR`, oldest deleted first; see [Report limits](#report-limits) |
| `DEBUG` | `false` | Enable debug logging | | `DEBUG` | `false` | Enable debug logging |
| `TRUSTED_PROXIES` | loopback + RFC1918 | Comma-separated CIDRs whose `X-Forwarded-For` / `X-Real-IP` headers are trusted for client IP resolution | | `TRUSTED_PROXIES` | loopback + RFC1918 | Comma-separated CIDRs whose `X-Forwarded-For` / `X-Real-IP` headers are trusted for client IP resolution |
| `REPORTS_PER_MINUTE` | `60` | Reports each client address may send a minute; see [Report limits](#report-limits) | | `REPORTS_PER_MINUTE` | `60` | Reports each client address may send a minute; see [Report limits](#report-limits) |
@@ -148,11 +148,17 @@ credentials, so it is bounded instead. Both refusals below answer with the same
`X-RateLimit-Reset` headers. `X-RateLimit-Reset` headers.
- **Size cap.** The report files in `DATA_DIR` may total at most - **Size cap.** The report files in `DATA_DIR` may total at most
`DATA_DIR_MAX_BYTES`, counting the files already there at start. Reports `DATA_DIR_MAX_BYTES`, counting the files already there at start. Reports
waiting in memory count at their uncompressed size until they are written, so waiting in memory count at their uncompressed size until they are written;
a report that would take the total past the cap is refused with 507, and those lost to a failed write stop counting, and the part of its file written
nothing of it is stored. Deleting report files frees room only at the next is removed. When a report would take the total past the cap, the oldest report
start, when the files are counted again. The default of 1 GiB is small enough files are deleted to make room, and each deletion is logged with the file's
for any host; set it to the space you can give `DATA_DIR`. name and size; a file still being written is never deleted. A report is
refused with 507, and nothing of it is stored, only when the reports waiting
to be written fill the cap on their own, and then no file is deleted. At
start, report files past the cap, as after lowering it, are deleted the same
way. So the cap is how much of the newest reports is kept: the default of 1
GiB is small enough for any host; set it to the space you can give
`DATA_DIR`.
### CORS ### CORS
@@ -170,7 +176,6 @@ starting, with an error naming `CORS_ALLOWED_ORIGINS`.
- Add integration test that POSTs a report and verifies the compressed output - Add integration test that POSTs a report and verifies the compressed output
- Add report decompression/query endpoint - Add report decompression/query endpoint
- Add metrics (Prometheus) for buffer size, flush count, report count - Add metrics (Prometheus) for buffer size, flush count, report count
- Add retention policy to prune old report files
## License ## License
+4 -3
View File
@@ -86,11 +86,12 @@ func (s *Handlers) decodeErrorStatus(err error) int {
} }
// appendErrorStatus logs a failure to store a report and returns // appendErrorStatus logs a failure to store a report and returns
// the status to send: 507 when the report files are at their size // the status to send: 507 when the reports waiting to be written fill
// cap, otherwise 500. // the size cap, otherwise 500.
func (s *Handlers) appendErrorStatus(err error) int { func (s *Handlers) appendErrorStatus(err error) int {
if errors.Is(err, reportbuf.ErrFull) { if errors.Is(err, reportbuf.ErrFull) {
s.log.Warn("report refused: report files at their size cap") s.log.Warn("report refused: " +
"reports waiting to be written fill the size cap")
return http.StatusInsufficientStorage return http.StatusInsufficientStorage
} }
+3 -3
View File
@@ -68,9 +68,9 @@ func TestHandleReportStorageFailureIsNon2xx(t *testing.T) {
} }
} }
// TestHandleReportFullIs507 checks the answer when the report files // TestHandleReportFullIs507 checks the answer when the reports waiting
// are at their size cap: 507 and the usual error body, which tells // to be written fill the size cap: 507 and the usual error body, which
// the client nothing more. // tells the client nothing more.
func TestHandleReportFullIs507(t *testing.T) { func TestHandleReportFullIs507(t *testing.T) {
t.Parallel() t.Parallel()
+11 -1
View File
@@ -1,6 +1,9 @@
package reportbuf package reportbuf
import "time" import (
"os"
"time"
)
// Flush writes the buffered reports to a file now, as the periodic // Flush writes the buffered reports to a file now, as the periodic
// flush does, so tests need not wait a minute for it. // flush does, so tests need not wait a minute for it.
@@ -13,3 +16,10 @@ func (b *Buffer) Flush() error {
func (b *Buffer) StopClock(at time.Time) { func (b *Buffer) StopClock(at time.Time) {
b.now = func() time.Time { return at } b.now = func() time.Time { return at }
} }
// OnFileCreated makes the buffer call fn with each report file it
// writes from now on, once the file is created and before anything is
// written to it.
func (b *Buffer) OnFileCreated(fn func(f *os.File)) {
b.fileCreated = fn
}
+156 -56
View File
@@ -1,5 +1,6 @@
// Package reportbuf accumulates telemetry reports in memory // Package reportbuf accumulates telemetry reports in memory
// and periodically flushes them to zstd-compressed JSONL files. // and periodically flushes them to zstd-compressed JSONL files,
// deleting the oldest files to keep them under a size cap.
package reportbuf package reportbuf
import ( import (
@@ -37,9 +38,10 @@ const (
fileSuffix = ".jsonl.zst" fileSuffix = ".jsonl.zst"
) )
// ErrFull is returned by Append when storing the report would // ErrFull is returned by Append when the reports waiting to be
// take the report files past the configured maximum size. // written leave no room for the report under the configured maximum
var ErrFull = errors.New("report files at their size cap") // size, however many report files are deleted.
var ErrFull = errors.New("reports waiting to be written fill the size cap")
// Params defines the dependencies for Buffer. // Params defines the dependencies for Buffer.
type Params struct { type Params struct {
@@ -49,15 +51,32 @@ type Params struct {
Logger *logger.Logger Logger *logger.Logger
} }
// reportFile is a report file that may be deleted to make room, with
// the size it counts for in usedBytes.
type reportFile struct {
name string
size int64
}
// Buffer accumulates JSON lines in memory and flushes them // Buffer accumulates JSON lines in memory and flushes them
// to zstd-compressed files on disk. // to zstd-compressed files on disk.
type Buffer struct { type Buffer struct {
buf bytes.Buffer buf bytes.Buffer
dataDir string dataDir string
done chan struct{} done chan struct{}
log *slog.Logger // fileCreated is called with each report file once it is
maxBytes int64 // created, before anything is written to it: it does nothing,
mu sync.Mutex // except in tests that hold the write open or make it fail.
fileCreated func(f *os.File)
// files are the report files that may be deleted to make room,
// oldest first: those in dataDir at start, then each one this
// buffer writes, once it is complete. A file still being written
// is not among them. filesBytes is their total size.
files []reportFile
filesBytes int64
log *slog.Logger
maxBytes int64
mu sync.Mutex
// now is the clock report files are named by: time.Now, except // now is the clock report files are named by: time.Now, except
// in tests that need two flushes to share a timestamp. // in tests that need two flushes to share a timestamp.
now func() time.Time now func() time.Time
@@ -66,8 +85,8 @@ type Buffer struct {
seq atomic.Uint64 seq atomic.Uint64
stopOnce sync.Once stopOnce sync.Once
// usedBytes is what Append checks against maxBytes: the size // usedBytes is what Append checks against maxBytes: the size
// of the report files in dataDir, plus the reports not yet // of the report files in dataDir, plus the reports waiting to
// written to one at their uncompressed size. // be written to one at their uncompressed size.
usedBytes int64 usedBytes int64
} }
@@ -83,11 +102,12 @@ func New(
} }
b := &Buffer{ b := &Buffer{
dataDir: dir, dataDir: dir,
done: make(chan struct{}), done: make(chan struct{}),
log: params.Logger.Get(), fileCreated: func(*os.File) {},
maxBytes: params.Config.DataDirMaxBytes, log: params.Logger.Get(),
now: time.Now, maxBytes: params.Config.DataDirMaxBytes,
now: time.Now,
} }
lc.Append(fx.Hook{ lc.Append(fx.Hook{
@@ -97,12 +117,28 @@ func New(
return fmt.Errorf("create data dir: %w", err) return fmt.Errorf("create data dir: %w", err)
} }
// Report files left by earlier runs count too. // Report files left by earlier runs count too, and are
b.usedBytes, err = reportFilesSize(b.dataDir) // the first to be deleted to make room.
files, err := reportFiles(b.dataDir)
if err != nil { if err != nil {
return err return err
} }
b.mu.Lock()
b.files = files
for _, f := range files {
b.filesBytes += f.size
}
b.usedBytes = b.filesBytes
// The files may be past the cap, if it was lowered since
// the last run.
b.deleteOldestFiles(0)
b.mu.Unlock()
go b.flushLoop() go b.flushLoop()
return nil return nil
@@ -128,9 +164,10 @@ func New(
} }
// Append marshals v as a single JSON line and appends it to // Append marshals v as a single JSON line and appends it to
// the buffer. It stores nothing and returns ErrFull if the line // the buffer. If the line would take usedBytes past maxBytes, the
// would take usedBytes past maxBytes. If the buffer reaches the // oldest report files are deleted to make room; it stores nothing
// size threshold, it is drained and written to disk // and returns ErrFull if that cannot make room. If the buffer
// reaches the size threshold, it is drained and written to disk
// asynchronously. // asynchronously.
func (b *Buffer) Append(v any) error { func (b *Buffer) Append(v any) error {
line, err := json.Marshal(v) line, err := json.Marshal(v)
@@ -142,6 +179,8 @@ func (b *Buffer) Append(v any) error {
b.mu.Lock() b.mu.Lock()
b.deleteOldestFiles(lineBytes)
if b.usedBytes+lineBytes > b.maxBytes { if b.usedBytes+lineBytes > b.maxBytes {
b.mu.Unlock() b.mu.Unlock()
@@ -171,6 +210,37 @@ func (b *Buffer) Append(v any) error {
return nil return nil
} }
// deleteOldestFiles deletes report files, oldest first, until n more
// bytes fit under maxBytes. It deletes none when the reports waiting
// to be written leave no room for n even with every file gone, since
// that would lose the files for nothing. The caller must hold b.mu.
func (b *Buffer) deleteOldestFiles(n int64) {
for b.usedBytes+n > b.maxBytes && len(b.files) > 0 {
if b.usedBytes-b.filesBytes+n > b.maxBytes {
return
}
f := b.files[0]
b.files = b.files[1:]
b.filesBytes -= f.size
// A file already gone, deleted by hand, has freed its room too.
err := os.Remove(filepath.Join(b.dataDir, f.name))
if err != nil && !errors.Is(err, fs.ErrNotExist) {
// The file is still there, so it still counts. It is
// not tried again until the next start.
b.log.Error("delete report file failed",
"file", f.name, "error", err)
continue
}
b.usedBytes -= f.size
b.log.Info("deleted report file to make room",
"file", f.name, "bytes", f.size)
}
}
// flushLoop runs a ticker that periodically flushes buffered // flushLoop runs a ticker that periodically flushes buffered
// data to disk until the done channel is closed. // data to disk until the done channel is closed.
func (b *Buffer) flushLoop() { func (b *Buffer) flushLoop() {
@@ -217,15 +287,41 @@ func (b *Buffer) drainBuf() []byte {
return data return data
} }
// writeFile creates a timestamped zstd-compressed JSONL file // writeFile writes data, reports drained from the buffer, to a new
// in the data directory. // timestamped zstd-compressed JSONL file in the data directory.
func (b *Buffer) writeFile(data []byte) error { func (b *Buffer) writeFile(data []byte) error {
// The timestamp comes first, so the names sort by time; the number // The timestamp comes first, so the names sort by time; the number
// after it tells apart files named in the same millisecond. // after it tells apart files named in the same millisecond.
ts := b.now().UTC().Format("2006-01-02T15-04-05.000Z") ts := b.now().UTC().Format("2006-01-02T15-04-05.000Z")
name := fmt.Sprintf("%s%s-%d%s", filePrefix, ts, b.seq.Add(1), fileSuffix) name := fmt.Sprintf("%s%s-%d%s", filePrefix, ts, b.seq.Add(1), fileSuffix)
path := filepath.Join(b.dataDir, name)
size, err := b.createFile(filepath.Join(b.dataDir, name), data)
// The reports no longer wait to be written, so they stop counting
// at their uncompressed size. If the write failed they are lost;
// otherwise they count as the file, which from here on may be
// deleted to make room.
b.mu.Lock()
defer b.mu.Unlock()
b.usedBytes -= int64(len(data))
if err != nil {
return err
}
b.usedBytes += size
b.files = append(b.files, reportFile{name: name, size: size})
b.filesBytes += size
return nil
}
// createFile creates the file at path holding data compressed with
// zstd, and returns its size. If the write fails once the file is
// created, it removes the file, so that a failed write leaves nothing
// behind to take room.
func (b *Buffer) createFile(path string, data []byte) (int64, error) {
// path is built from the operator-supplied dataDir plus a // path is built from the operator-supplied dataDir plus a
// generated timestamp and number, so it carries no external input. // generated timestamp and number, so it carries no external input.
f, err := os.OpenFile( //nolint:gosec // see comment above f, err := os.OpenFile( //nolint:gosec // see comment above
@@ -234,61 +330,65 @@ func (b *Buffer) writeFile(data []byte) error {
filePerms, filePerms,
) )
if err != nil { if err != nil {
return fmt.Errorf("create report file: %w", err) return 0, fmt.Errorf("create report file: %w", err)
} }
// Closes the file on the early returns below. The success b.fileCreated(f)
// path closes it explicitly to check the error; closing it
// a second time here is harmless.
defer func() { _ = f.Close() }()
size, err := writeCompressed(f, data)
if err != nil {
_ = f.Close()
return 0, errors.Join(err, os.Remove(path))
}
err = f.Close()
if err != nil {
err = fmt.Errorf("close report file: %w", err)
return 0, errors.Join(err, os.Remove(path))
}
return size, nil
}
// writeCompressed writes data to f compressed with zstd, and returns
// the size of f.
func writeCompressed(f *os.File, data []byte) (int64, error) {
enc, err := zstd.NewWriter(f) enc, err := zstd.NewWriter(f)
if err != nil { if err != nil {
return fmt.Errorf("create zstd encoder: %w", err) return 0, fmt.Errorf("create zstd encoder: %w", err)
} }
_, err = enc.Write(data) _, err = enc.Write(data)
if err != nil { if err != nil {
_ = enc.Close() _ = enc.Close()
return fmt.Errorf("write compressed data: %w", err) return 0, fmt.Errorf("write compressed data: %w", err)
} }
err = enc.Close() err = enc.Close()
if err != nil { if err != nil {
return fmt.Errorf("close zstd encoder: %w", err) return 0, fmt.Errorf("close zstd encoder: %w", err)
} }
info, err := f.Stat() info, err := f.Stat()
if err != nil { if err != nil {
return fmt.Errorf("stat report file: %w", err) return 0, fmt.Errorf("stat report file: %w", err)
} }
err = f.Close() return info.Size(), nil
if err != nil {
return fmt.Errorf("close report file: %w", err)
}
// The reports counted at their uncompressed size while they
// waited; now they count as the file. After a failed write they
// stay counted as they were, which errs toward refusing reports
// early rather than letting the files pass the cap.
b.mu.Lock()
b.usedBytes += info.Size() - int64(len(data))
b.mu.Unlock()
return nil
} }
// reportFilesSize returns the total size of the report files in // reportFiles returns the report files in dir, oldest first:
// dir. // os.ReadDir sorts them by name, and the names sort by time.
func reportFilesSize(dir string) (int64, error) { func reportFiles(dir string) ([]reportFile, error) {
entries, err := os.ReadDir(dir) entries, err := os.ReadDir(dir)
if err != nil { if err != nil {
return 0, fmt.Errorf("read data dir: %w", err) return nil, fmt.Errorf("read data dir: %w", err)
} }
var total int64 files := make([]reportFile, 0, len(entries))
for _, entry := range entries { for _, entry := range entries {
name := entry.Name() name := entry.Name()
@@ -299,11 +399,11 @@ func reportFilesSize(dir string) (int64, error) {
info, err := entry.Info() info, err := entry.Info()
if err != nil { if err != nil {
return 0, fmt.Errorf("stat report file: %w", err) return nil, fmt.Errorf("stat report file: %w", err)
} }
total += info.Size() files = append(files, reportFile{name: name, size: info.Size()})
} }
return total, nil return files, nil
} }
+327 -34
View File
@@ -139,11 +139,19 @@ func lineBytes(t *testing.T, report any) int {
return len(line) + 1 return len(line) + 1
} }
// TestAppendPastCapIsRefused fills the cap with a report not yet
// written. The next report is refused, and the report file already in
// DATA_DIR is kept: it is smaller than a report, so deleting it could
// not make room.
func TestAppendPastCapIsRefused(t *testing.T) { func TestAppendPastCapIsRefused(t *testing.T) {
report := map[string]string{"id": "cap"} report := map[string]string{"id": "cap"}
dir := t.TempDir()
earlier := reportFilePath(dir, 1)
t.Setenv("DATA_DIR", t.TempDir()) writeBytes(t, earlier, 1)
t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(lineBytes(t, report)))
t.Setenv("DATA_DIR", dir)
t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(1+lineBytes(t, report)))
buf := startBuffer(t) buf := startBuffer(t)
@@ -156,24 +164,38 @@ func TestAppendPastCapIsRefused(t *testing.T) {
if !errors.Is(err, reportbuf.ErrFull) { if !errors.Is(err, reportbuf.ErrFull) {
t.Fatalf("report past the cap: error = %v, want ErrFull", err) t.Fatalf("report past the cap: error = %v, want ErrFull", err)
} }
if !exists(t, earlier) {
t.Fatal("report file deleted, though that could not make room")
}
} }
// TestCapCountsReportFilesAlreadyInDataDir starts on a data // TestOldestReportFileDeletedFirst starts on a data directory holding
// directory holding a report file from an earlier run, and a file // report files from an earlier run, and a file that is not a report,
// that is not a report, which must not count. // which neither counts nor is ever deleted. Nothing is deleted while
func TestCapCountsReportFilesAlreadyInDataDir(t *testing.T) { // there is room; then only the oldest report file is.
const earlierBytes = 100 func TestOldestReportFileDeletedFirst(t *testing.T) {
const fileBytes = 100
report := map[string]string{"id": "cap"} report := map[string]string{"id": "oldest"}
dir := t.TempDir() dir := t.TempDir()
oldest := reportFilePath(dir, 1)
kept := []string{
reportFilePath(dir, 2),
reportFilePath(dir, 3),
filepath.Join(dir, "notes.txt"),
}
writeBytes(t, filepath.Join(dir, "reports-2026-01-01T00-00-00.000Z.jsonl.zst"), writeBytes(t, oldest, fileBytes)
earlierBytes)
writeBytes(t, filepath.Join(dir, "notes.txt"), 10*earlierBytes) for _, path := range kept {
writeBytes(t, path, fileBytes)
}
t.Setenv("DATA_DIR", dir) t.Setenv("DATA_DIR", dir)
// Room for the three report files and one report.
t.Setenv("DATA_DIR_MAX_BYTES", t.Setenv("DATA_DIR_MAX_BYTES",
strconv.Itoa(earlierBytes+lineBytes(t, report))) strconv.Itoa(3*fileBytes+lineBytes(t, report)))
buf := startBuffer(t) buf := startBuffer(t)
@@ -182,9 +204,55 @@ func TestCapCountsReportFilesAlreadyInDataDir(t *testing.T) {
t.Fatalf("report that fills the cap exactly: %v", err) t.Fatalf("report that fills the cap exactly: %v", err)
} }
if !exists(t, oldest) {
t.Fatal("oldest report file deleted while there was room")
}
err = buf.Append(report) err = buf.Append(report)
if !errors.Is(err, reportbuf.ErrFull) { if err != nil {
t.Fatalf("report past the cap: error = %v, want ErrFull", err) t.Fatalf("report past the cap: %v", err)
}
if exists(t, oldest) {
t.Fatal("oldest report file kept when room was needed")
}
for _, path := range kept {
if !exists(t, path) {
t.Fatalf("%s deleted; only the oldest report file should be", path)
}
}
}
// TestStartDeletesFilesPastCap starts on report files past the cap,
// as after the cap is lowered: the oldest are deleted until the rest
// fit.
func TestStartDeletesFilesPastCap(t *testing.T) {
const fileBytes = 100
dir := t.TempDir()
oldest := reportFilePath(dir, 1)
kept := []string{reportFilePath(dir, 2), reportFilePath(dir, 3)}
writeBytes(t, oldest, fileBytes)
for _, path := range kept {
writeBytes(t, path, fileBytes)
}
t.Setenv("DATA_DIR", dir)
t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(len(kept)*fileBytes))
startBuffer(t)
if exists(t, oldest) {
t.Fatal("oldest report file kept, though the files were past the cap")
}
for _, path := range kept {
if !exists(t, path) {
t.Fatalf("%s deleted, though the rest fit without it", path)
}
} }
} }
@@ -195,8 +263,9 @@ func TestWrittenReportsCountAtFileSize(t *testing.T) {
// Repetitive, so its file is far smaller than its JSON. // Repetitive, so its file is far smaller than its JSON.
report := map[string]string{"id": strings.Repeat("a", 1000)} report := map[string]string{"id": strings.Repeat("a", 1000)}
size := lineBytes(t, report) size := lineBytes(t, report)
dir := t.TempDir()
t.Setenv("DATA_DIR", t.TempDir()) t.Setenv("DATA_DIR", dir)
// Room for the report twice over only if the first one counts // Room for the report twice over only if the first one counts
// at its file's size by the time the second arrives. // at its file's size by the time the second arrives.
t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(2*size-1)) t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(2*size-1))
@@ -217,13 +286,22 @@ func TestWrittenReportsCountAtFileSize(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("second report, after the first was written: %v", err) t.Fatalf("second report, after the first was written: %v", err)
} }
if !hasReportFile(t, dir) {
t.Fatal("first report's file deleted, though the second fit beside it")
}
} }
// TestWrittenReportsKeepCounting writes one report file after another // TestWrittenFilesDeletedToMakeRoom writes one report file after
// under a small cap: each report must be taken while the files on disk // another under a small cap. Every report must be taken; files are
// leave room for it, and refused once they do not. // deleted only when the report would not fit beside them, and the
func TestWrittenReportsKeepCounting(t *testing.T) { // files kept leave room for it.
const maxBytes = 200 func TestWrittenFilesDeletedToMakeRoom(t *testing.T) {
const (
maxBytes = 200
// One file each, which take far more than maxBytes together.
reports = 50
)
report := map[string]string{"id": "written"} report := map[string]string{"id": "written"}
size := int64(lineBytes(t, report)) size := int64(lineBytes(t, report))
@@ -234,23 +312,24 @@ func TestWrittenReportsKeepCounting(t *testing.T) {
buf := startBuffer(t) buf := startBuffer(t)
// Every file takes at least a byte, so they fill the cap within for range reports {
// maxBytes rounds. before := reportFilesBytes(t, dir)
for range maxBytes {
used := reportFilesBytes(t, dir)
err := buf.Append(report) err := buf.Append(report)
if used+size > maxBytes { if err != nil {
if !errors.Is(err, reportbuf.ErrFull) { t.Fatalf("with %d bytes of report files: %v", before, err)
t.Fatalf("with %d bytes of report files: error = %v, "+
"want ErrFull", used, err)
}
return
} }
if err != nil { after := reportFilesBytes(t, dir)
t.Fatalf("with %d bytes of report files: %v", used, err)
if before+size <= maxBytes && after != before {
t.Fatalf("files deleted, though the report fit beside "+
"their %d bytes", before)
}
if after+size > maxBytes {
t.Fatalf("%d bytes of report files kept, leaving no room "+
"for the report", after)
} }
err = buf.Flush() err = buf.Flush()
@@ -258,8 +337,199 @@ func TestWrittenReportsKeepCounting(t *testing.T) {
t.Fatalf("flush: %v", err) t.Fatalf("flush: %v", err)
} }
} }
}
t.Fatal("the report files never filled the cap") // TestFailedDeletionStillCounts makes deleting the oldest report file
// fail. It is still there, so it still takes room, and the next oldest
// is deleted in its place.
func TestFailedDeletionStillCounts(t *testing.T) {
const fileBytes = 100
report := map[string]string{"id": "stuck"}
dir := t.TempDir()
stuck := reportFilePath(dir, 1)
next := reportFilePath(dir, 2)
newest := reportFilePath(dir, 3)
for _, path := range []string{stuck, next, newest} {
writeBytes(t, path, fileBytes)
}
t.Setenv("DATA_DIR", dir)
t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(3*fileBytes))
buf := startBuffer(t)
// A directory that is not empty cannot be deleted, even by root,
// which the tests run as in the backend image.
err := os.Remove(stuck)
if err != nil {
t.Fatalf("remove %s: %v", stuck, err)
}
err = os.Mkdir(stuck, 0o750)
if err != nil {
t.Fatalf("make directory %s: %v", stuck, err)
}
writeBytes(t, filepath.Join(stuck, "file"), 1)
err = buf.Append(report)
if err != nil {
t.Fatalf("report past the cap: %v", err)
}
if exists(t, next) {
t.Fatal("next oldest report file kept: the failed deletion " +
"counted as making room")
}
if !exists(t, newest) {
t.Fatal("newest report file deleted, though deleting one made room")
}
}
// TestFileDeletedByHandFreesRoom deletes the oldest report file by
// hand after start. When room is needed, its room counts as freed, so
// no other file is deleted.
func TestFileDeletedByHandFreesRoom(t *testing.T) {
const fileBytes = 100
report := map[string]string{"id": "by-hand"}
dir := t.TempDir()
gone := reportFilePath(dir, 1)
kept := reportFilePath(dir, 2)
writeBytes(t, gone, fileBytes)
writeBytes(t, kept, fileBytes)
t.Setenv("DATA_DIR", dir)
t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(2*fileBytes))
buf := startBuffer(t)
err := os.Remove(gone)
if err != nil {
t.Fatalf("remove %s: %v", gone, err)
}
err = buf.Append(report)
if err != nil {
t.Fatalf("report past the cap: %v", err)
}
if !exists(t, kept) {
t.Fatal("report file deleted, though the one deleted by hand " +
"had made room")
}
}
// TestFileBeingWrittenIsNeverDeleted holds a write open, its file
// created but not complete, while a report needs room. The file may
// not be deleted to make it, so the report is refused. Once the write
// is complete, the file is deleted when room is needed.
func TestFileBeingWrittenIsNeverDeleted(t *testing.T) {
report := map[string]string{"id": "writing"}
t.Setenv("DATA_DIR", t.TempDir())
t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(lineBytes(t, report)))
buf := startBuffer(t)
created := make(chan string)
release := make(chan struct{})
buf.OnFileCreated(func(f *os.File) {
created <- f.Name()
<-release
})
err := buf.Append(report)
if err != nil {
t.Fatalf("report that fills the cap exactly: %v", err)
}
flushed := make(chan error)
go func() { flushed <- buf.Flush() }()
writing := <-created
// Errorf, not Fatalf, until the write is released, so that a
// failure here does not leave it held.
err = buf.Append(report)
if !errors.Is(err, reportbuf.ErrFull) {
t.Errorf("report while the first was being written: "+
"error = %v, want ErrFull", err)
}
if !exists(t, writing) {
t.Error("report file deleted while it was being written")
}
close(release)
err = <-flushed
if err != nil {
t.Fatalf("flush: %v", err)
}
// Writes from here on, the final one at stop included, go
// through unheld.
buf.OnFileCreated(func(*os.File) {})
err = buf.Append(report)
if err != nil {
t.Fatalf("report after the write was complete: %v", err)
}
if exists(t, writing) {
t.Fatal("complete report file kept when room was needed")
}
}
// TestFailedWriteStopsCounting makes a write fail once its file is
// created. Its reports are lost, so they stop counting, and the part of
// the file written is removed, so it takes no room.
func TestFailedWriteStopsCounting(t *testing.T) {
report := map[string]string{"id": "failed"}
t.Setenv("DATA_DIR", t.TempDir())
t.Setenv("DATA_DIR_MAX_BYTES", strconv.Itoa(lineBytes(t, report)))
buf := startBuffer(t)
var failed string
// Closing the file under the write makes the write fail.
buf.OnFileCreated(func(f *os.File) {
failed = f.Name()
_ = f.Close()
})
err := buf.Append(report)
if err != nil {
t.Fatalf("report that fills the cap exactly: %v", err)
}
err = buf.Flush()
if err == nil {
t.Fatal("flush succeeded, though its file was closed under it")
}
if exists(t, failed) {
t.Fatal("file of the failed write kept")
}
// Writes from here on, the final one at stop included, succeed.
buf.OnFileCreated(func(*os.File) {})
err = buf.Append(report)
if err != nil {
t.Fatalf("report after the failed write: %v", err)
}
} }
// TestConcurrentAppendsStopAtCap appends from many goroutines at once // TestConcurrentAppendsStopAtCap appends from many goroutines at once
@@ -407,6 +677,14 @@ func readReportFiles(t *testing.T, dir string) []string {
return contents return contents
} }
// reportFilePath returns the path in dir of a report file named as
// written on the given day of January 2026, so that a lower day sorts
// as older.
func reportFilePath(dir string, day int) string {
return filepath.Join(dir,
fmt.Sprintf("reports-2026-01-%02dT00-00-00.000Z-1.jsonl.zst", day))
}
func writeBytes(t *testing.T, path string, n int) { func writeBytes(t *testing.T, path string, n int) {
t.Helper() t.Helper()
@@ -416,6 +694,21 @@ func writeBytes(t *testing.T, path string, n int) {
} }
} }
func exists(t *testing.T, path string) bool {
t.Helper()
_, err := os.Stat(path)
if errors.Is(err, fs.ErrNotExist) {
return false
}
if err != nil {
t.Fatalf("stat %s: %v", path, err)
}
return true
}
func hasReportFile(t *testing.T, dir string) bool { func hasReportFile(t *testing.T, dir string) bool {
t.Helper() t.Helper()