amadan

carlos / platform Public

analytics: atomic salt file; a corrupt visitors sketch is replaced, not fatal

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 3cfb076c6bde3edeadac63f5d69bbfb32b14742c parent f763a2b
4 files changed, +113 −11
  • internal/analytics/day.go +7 −4
  • internal/analytics/day_test.go +18 −0
  • internal/analytics/salt.go +57 −7
  • internal/analytics/salt_test.go +31 −0
diff --git a/internal/analytics/day.go b/internal/analytics/day.go
index 14614b3d0..0ae9ab2fd 100644
--- a/internal/analytics/day.go
+++ b/internal/analytics/day.go
@@ -75,12 +75,12 @@ func DayKey(sqid, app, box, date string) string { return BoxPrefix(sqid, app, bo
// Merge folds x into d unless x.Seq is not above the recorded seq for
// x.Box — the idempotency rule that makes a retried conditional put safe.
func (d *Day) Merge(x Delta) (bool, error) {
- if d.Seq == nil {
- d.Seq = map[string]int64{}
- }
if x.Seq <= d.Seq[x.Box] {
return false, nil
}
+ if d.Seq == nil {
+ d.Seq = map[string]int64{}
+ }
d.V, d.Date = 1, x.Date
if d.Hours == nil {
d.Hours = map[string]HourRow{}
@@ -92,9 +92,12 @@ func (d *Day) Merge(x Delta) (bool, error) {
cur.Bytes += row.Bytes
d.Hours[h] = cur
}
+ // A corrupt or unreadable sketch must not poison the whole day: start
+ // fresh rather than fail the merge. The bad string is overwritten by
+ // the freshly encoded sketch below either way.
vis, err := DecodeHLL(d.Visitors)
if err != nil {
- return false, err
+ vis = &HLL{}
}
if x.Visitors != nil {
vis.Merge(x.Visitors)
diff --git a/internal/analytics/day_test.go b/internal/analytics/day_test.go
index 2990a0479..f702d4000 100644
--- a/internal/analytics/day_test.go
+++ b/internal/analytics/day_test.go
@@ -47,6 +47,24 @@ func TestDayMergeIsSeqIdempotent(t *testing.T) {
}
}
+func TestDayMergeReplacesCorruptSketch(t *testing.T) {
+ d := Day{Visitors: "garbage!"}
+ ok, err := d.Merge(delta("box-a", 1, 5))
+ if !ok || err != nil {
+ t.Fatalf("merge with corrupt sketch: %v %v", ok, err)
+ }
+ if d.Hours["09"].PV != 5 || d.Pages["/"] != 5 {
+ t.Fatalf("day = %+v", d)
+ }
+ vis, err := DecodeHLL(d.Visitors)
+ if err != nil {
+ t.Fatalf("decode replaced sketch: %v", err)
+ }
+ if c := vis.Count(); c < 0.5 {
+ t.Fatalf("expected at least one visitor recorded, got %.2f", c)
+ }
+}
+
func TestUpsertDayCASAcrossBoxes(t *testing.T) {
ctx := context.Background()
s := release.NewDirStore(t.TempDir())
diff --git a/internal/analytics/salt.go b/internal/analytics/salt.go
index 8272958fe..812407658 100644
--- a/internal/analytics/salt.go
+++ b/internal/analytics/salt.go
@@ -2,45 +2,95 @@ package analytics
import (
"crypto/rand"
+ "encoding/hex"
"errors"
+ "fmt"
"os"
"path/filepath"
"strings"
+ "time"
)
const saltPrefix = "analytics-salt-"
+const tmpInfix = ".tmp-"
// DaySalt returns the box's visitor salt for date (YYYY-MM-DD), creating
// <dir>/analytics-salt-<date> with 32 random bytes at 0600 when absent and
// deleting any other day's file (spec §3.3: restart-stable within the day,
// gone on rotation, never in the bucket). dir is the box data directory.
+//
+// The file is created atomically so concurrent callers never observe a
+// short or empty salt while another goroutine/process is still writing:
+// a candidate is written in full to a private
+// analytics-salt-<date>.tmp-<random> file first, then linked into place
+// with os.Link. Link fails with os.ErrExist when a sibling has already
+// won the race; the loser then polls the winner's file for up to 200ms
+// until it has settled to 32 bytes, rather than trust a possibly-partial
+// read. Filesystems without hard-link support (some Windows
+// configurations) fall back to os.Rename, which is atomic but — unlike
+// Link — can clobber a concurrent winner's file; that's an accepted
+// trade-off on those filesystems.
func DaySalt(dir, date string) ([]byte, error) {
path := filepath.Join(dir, saltPrefix+date)
if b, err := os.ReadFile(path); err == nil && len(b) == 32 {
return b, nil
}
+
b := make([]byte, 32)
if _, err := rand.Read(b); err != nil {
return nil, err
}
- f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
+
+ var suffix [8]byte
+ if _, err := rand.Read(suffix[:]); err != nil {
+ return nil, err
+ }
+ tmp := path + tmpInfix + hex.EncodeToString(suffix[:])
+
+ f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
if err != nil {
- if errors.Is(err, os.ErrExist) { // a sibling process won the race
- return os.ReadFile(path)
- }
return nil, err
}
_, werr := f.Write(b)
cerr := f.Close()
if werr != nil || cerr != nil {
- os.Remove(path)
+ os.Remove(tmp)
return nil, errors.Join(werr, cerr)
}
+
+ switch linkErr := os.Link(tmp, path); {
+ case linkErr == nil:
+ // We won: tmp and path now name the same inode.
+ os.Remove(tmp)
+ case errors.Is(linkErr, os.ErrExist):
+ // A sibling process/goroutine won the race.
+ os.Remove(tmp)
+ for i := 0; i < 20; i++ {
+ if got, rerr := os.ReadFile(path); rerr == nil && len(got) == 32 {
+ return got, nil
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ return nil, fmt.Errorf("analytics: timed out waiting for %s", path)
+ default:
+ // No hard-link support: fall back to an atomic rename.
+ if rerr := os.Rename(tmp, path); rerr != nil {
+ os.Remove(tmp)
+ return nil, rerr
+ }
+ }
+
// Rotation: every other day's salt goes, so nothing on disk can join a
- // hash across days.
+ // hash across days. In-flight tmpInfix files are skipped — they also
+ // share saltPrefix (possibly even today's date, from a sibling
+ // goroutine/process still mid-write) and must not be reaped out from
+ // under their owner. A failed ReadDir here is deliberately ignored:
+ // dir is local per-box state, so a stale salt file left behind
+ // self-heals the next time rotation runs.
entries, _ := os.ReadDir(dir)
for _, e := range entries {
- if n := e.Name(); strings.HasPrefix(n, saltPrefix) && n != saltPrefix+date {
+ n := e.Name()
+ if strings.HasPrefix(n, saltPrefix) && n != saltPrefix+date && !strings.Contains(n, tmpInfix) {
os.Remove(filepath.Join(dir, n))
}
}
diff --git a/internal/analytics/salt_test.go b/internal/analytics/salt_test.go
index d4f4a5ea9..90dfa25c1 100644
--- a/internal/analytics/salt_test.go
+++ b/internal/analytics/salt_test.go
@@ -4,6 +4,7 @@ import (
"os"
"path/filepath"
"runtime"
+ "sync"
"testing"
)
@@ -31,3 +32,33 @@ func TestDaySaltIsStableAndRotates(t *testing.T) {
t.Fatal("rotation must delete the previous day's salt")
}
}
+
+func TestDaySaltConcurrentCallersAgree(t *testing.T) {
+ dir := t.TempDir()
+ const n = 8
+ results := make([][]byte, n)
+ errs := make([]error, n)
+ var wg sync.WaitGroup
+ wg.Add(n)
+ for i := 0; i < n; i++ {
+ i := i
+ go func() {
+ defer wg.Done()
+ results[i], errs[i] = DaySalt(dir, "2026-09-07")
+ }()
+ }
+ wg.Wait()
+ for i, err := range errs {
+ if err != nil {
+ t.Fatalf("goroutine %d: %v", i, err)
+ }
+ if len(results[i]) != 32 {
+ t.Fatalf("goroutine %d: len=%d", i, len(results[i]))
+ }
+ }
+ for i := 1; i < n; i++ {
+ if string(results[i]) != string(results[0]) {
+ t.Fatalf("goroutine %d's salt disagrees with goroutine 0's", i)
+ }
+ }
+}