spb/internetpressure

1// Package spool persists gzip-compressed batches that could not be delivered, under data_dir/spool/.2//3// Files are named <unixnano>.json.gz so lexical order is chronological; the oldest is re-sent first. A size cap4// (default 200 MiB) drops the oldest files when exceeded. Files hold the exact compressed bytes: the signature5// covers those bytes, so each retry only needs a fresh timestamp.6package spool78import (9	"errors"10	"fmt"11	"os"12	"path/filepath"13	"sort"14	"strconv"15	"strings"16	"sync"17	"time"18)1920// DefaultCap is the on-disk size limit.21const DefaultCap int64 = 200 << 202223// Spool is a directory of pending batches. Safe for concurrent use.24type Spool struct {25	dir string26	cap int6427	mu  sync.Mutex28}2930// Open creates dir (0700) if needed. capBytes ≤ 0 selects DefaultCap.31func Open(dir string, capBytes int64) (*Spool, error) {32	if err := os.MkdirAll(dir, 0o700); err != nil {33		return nil, fmt.Errorf("spool: %w", err)34	}35	if capBytes <= 0 {36		capBytes = DefaultCap37	}38	return &Spool{dir: dir, cap: capBytes}, nil39}4041// Dir returns the spool directory.42func (s *Spool) Dir() string { return s.dir }4344// Write stores a compressed batch and enforces the cap. Returns the file name.45func (s *Spool) Write(gz []byte) (string, error) {46	s.mu.Lock()47	defer s.mu.Unlock()48	name := strconv.FormatInt(time.Now().UnixNano(), 10) + ".json.gz"49	final := filepath.Join(s.dir, name)50	tmp := final + ".tmp"51	if err := os.WriteFile(tmp, gz, 0o600); err != nil {52		return "", fmt.Errorf("spool: %w", err)53	}54	if err := os.Rename(tmp, final); err != nil {55		os.Remove(tmp)56		return "", fmt.Errorf("spool: %w", err)57	}58	s.enforceCapLocked()59	return name, nil60}6162// entry is one spooled file.63type entry struct {64	name string65	size int6466}6768func (s *Spool) listLocked() []entry {69	des, err := os.ReadDir(s.dir)70	if err != nil {71		return nil72	}73	out := make([]entry, 0, len(des))74	for _, de := range des {75		if de.IsDir() || !strings.HasSuffix(de.Name(), ".json.gz") {76			continue77		}78		info, err := de.Info()79		if err != nil {80			continue81		}82		out = append(out, entry{name: de.Name(), size: info.Size()})83	}84	sort.Slice(out, func(i, j int) bool { return out[i].name < out[j].name })85	return out86}8788func (s *Spool) enforceCapLocked() {89	entries := s.listLocked()90	var total int6491	for _, e := range entries {92		total += e.size93	}94	for i := 0; total > s.cap && i < len(entries); i++ {95		if err := os.Remove(filepath.Join(s.dir, entries[i].name)); err == nil {96			total -= entries[i].size97		}98	}99	// Stale temp files from a crash mid-write.100	des, _ := os.ReadDir(s.dir)101	for _, de := range des {102		if strings.HasSuffix(de.Name(), ".tmp") {103			os.Remove(filepath.Join(s.dir, de.Name()))104		}105	}106}107108// ErrEmpty is returned by Oldest when nothing is spooled.109var ErrEmpty = errors.New("spool: empty")110111// Oldest returns the oldest pending batch (name and compressed bytes).112func (s *Spool) Oldest() (string, []byte, error) {113	s.mu.Lock()114	defer s.mu.Unlock()115	for _, e := range s.listLocked() {116		data, err := os.ReadFile(filepath.Join(s.dir, e.name))117		if err != nil {118			os.Remove(filepath.Join(s.dir, e.name)) // unreadable → discard119			continue120		}121		return e.name, data, nil122	}123	return "", nil, ErrEmpty124}125126// Remove deletes a delivered (or rejected) batch.127func (s *Spool) Remove(name string) error {128	s.mu.Lock()129	defer s.mu.Unlock()130	err := os.Remove(filepath.Join(s.dir, filepath.Base(name)))131	if errors.Is(err, os.ErrNotExist) {132		return nil133	}134	return err135}136137// Stats returns (file count, total bytes).138func (s *Spool) Stats() (int, int64) {139	s.mu.Lock()140	defer s.mu.Unlock()141	entries := s.listLocked()142	var total int64143	for _, e := range entries {144		total += e.size145	}146	return len(entries), total147}148149// Bytes returns the total spooled size.150func (s *Spool) Bytes() int64 {151	_, b := s.Stats()152	return b153}154