amadan

eleven / ocho Public

server: blob mirror + audit — the second copy on a second provider (federated blobs Phase A, #992) (#1010)

Every store-backed blob gets a copy on a second S3-compatible store, and
both stores are spot-checked against the sha256 the upload now records.
Write-once ciphertext keeps it boring: the mirror sweep is idempotent and
order-free (the blobs table IS the journal), so any crash re-runs to
convergence.

- blobs table: sha256 (recorded while the upload streams — one pass) +
  mirrored_at columns, additive
- LCHAT_BLOB_S3_MIRROR_* env sextet; per-store static key pairs (primary and
  mirror are different providers — the AWS default chain only holds one);
  mirror-without-primary refused at boot
- mirror sweep (5m tick + an upload kick): copies what the mirror lacks,
  hashing in transit — pre-hash rows get backfilled, and a primary whose
  bytes stopped matching is NEVER mirrored (corruption must not spread to
  what may be the last good copy) and shouts
- reads chain primary → mirror → disk (a mirror serve after a primary miss
  logs loudly — that's redundancy paying out, and a healing signal); deletes
  reap both stores
- hourly audit full-reads a small sample from EACH store and verifies
  against the recorded hash — probabilistic coverage at a few requests per
  pass, results returned (not just logged) for Phase B's status publishing
- legacy in-DB rows excluded by construction (their bytes ride the DB's own
  replication)

Ships dark like #987: nothing changes until both env quartets are set. Live
flip = Hetzner Object Storage primary + the existing AWS bucket as mirror
(decisions 1-2 in docs/federated-blob-storage.md). Guide env rows updated.
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 08512d0e71e2256f82f0770907c6275e151086b7 parent 69d484a
7 files changed, +614 −27
  • blobmirror.go +291 −0
  • blobmirror_test.go +236 −0
  • db.go +31 −6
  • docs/planet-scale-guide/reference.html +2 −1
  • main.go +13 −3
  • s3blobstore.go +40 −16
  • s3blobstore_test.go +1 −1
diff --git a/blobmirror.go b/blobmirror.go
new file mode 100644
index 0000000..3414f8a
--- /dev/null
+++ b/blobmirror.go
@@ -0,0 +1,291 @@
+package main
+
+// 🤖 Blob mirroring + audit (federated blobs Phase A, #992 —
+// docs/federated-blob-storage.md): a second S3-compatible store holds a copy
+// of every store-backed blob, and both stores are spot-checked against the
+// sha256 the upload recorded. Write-once ciphertext makes all of this
+// embarrassingly simple: the mirror sweep is idempotent and order-free (the
+// blobs table IS the journal — no state machine, no lease), and a re-run
+// after any crash converges.
+//
+// Enabled per instance by the LCHAT_BLOB_S3_MIRROR_* env quartet (+ key
+// pair) beside the primary's; requires the primary S3 store (a mirror of the
+// bare disk layout is the fleet backup's job, not this one's). Reads chain
+// primary → mirror → disk, so a store outage degrades to slower reads, never
+// errors. Deletes reap both stores. Changing the mirror TARGET later means
+// resetting the watermark: UPDATE blobs SET mirrored_at=0 — documented beside
+// the env reference; Phase B's store registry replaces this with per-store
+// placement rows.
+
+import (
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "io"
+ "io/fs"
+ "log"
+ "time"
+)
+
+// mirroredBlobStore composes the primary (which itself falls back to disk for
+// pre-flip blobs) with the mirror. The store interface stays the seam: the
+// sweep below copies through create/commit, so nothing here knows S3.
+type mirroredBlobStore struct {
+ primary blobStore
+ mirror blobStore
+}
+
+func (m *mirroredBlobStore) create() (blobWriter, error) { return m.primary.create() }
+
+func (m *mirroredBlobStore) open(id string) (io.ReadSeekCloser, int64, time.Time, error) {
+ r, size, mod, err := m.primary.open(id)
+ if err == nil {
+ return r, size, mod, nil
+ }
+ if errors.Is(err, fs.ErrNotExist) {
+ // Primary outage looks like errors, not not-founds; a genuine miss
+ // (a lost primary object) served from the mirror is the redundancy
+ // paying out — loudly, because it means the primary needs healing.
+ if r2, s2, m2, err2 := m.mirror.open(id); err2 == nil {
+ log.Printf("blobs: %s served from the MIRROR (primary miss — investigate)", id)
+ return r2, s2, m2, nil
+ }
+ }
+ return nil, 0, time.Time{}, err
+}
+
+func (m *mirroredBlobStore) remove(id string) error {
+ err := m.primary.remove(id)
+ if err2 := m.mirror.remove(id); err == nil {
+ err = err2
+ }
+ return err
+}
+
+// ---- the mirror sweep -------------------------------------------------------
+
+// unmirroredBlobs returns store-backed rows the mirror lacks, oldest first.
+// Legacy in-DB rows (bytes in the data column) are excluded by the
+// length(data)=0 predicate — they ride the DB's own replication.
+func (db *DB) unmirroredBlobs(limit int) ([]struct{ ID, SHA string }, error) {
+ rows, err := db.sql.Query(
+ `SELECT id, sha256 FROM blobs
+ WHERE mirrored_at = 0 AND length(data) = 0
+ ORDER BY created_at LIMIT ?`, limit)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []struct{ ID, SHA string }
+ for rows.Next() {
+ var r struct{ ID, SHA string }
+ if err := rows.Scan(&r.ID, &r.SHA); err != nil {
+ return nil, err
+ }
+ out = append(out, r)
+ }
+ return out, rows.Err()
+}
+
+func (db *DB) markMirrored(id, sha256Hex string) error {
+ _, err := db.sql.Exec(
+ "UPDATE blobs SET mirrored_at = ?, sha256 = ? WHERE id = ?",
+ time.Now().Unix(), sha256Hex, id)
+ return err
+}
+
+// mirrorOnce copies one batch of unmirrored blobs primary→mirror, hashing in
+// transit: rows without a sha256 (pre-hash uploads) get it backfilled — the
+// sweep reads the full bytes anyway — and rows WITH one are verified, so a
+// primary that corrupted a blob is caught before the corruption spreads to
+// the copy. Returns how many blobs were mirrored this pass.
+func (db *DB) mirrorOnce(batch int) (int, error) {
+ m, ok := db.blobs.(*mirroredBlobStore)
+ if !ok {
+ return 0, nil
+ }
+ pending, err := db.unmirroredBlobs(batch)
+ if err != nil {
+ return 0, err
+ }
+ done := 0
+ for _, row := range pending {
+ r, _, _, err := m.primary.open(row.ID)
+ if err != nil {
+ if errors.Is(err, fs.ErrNotExist) {
+ // Deleted since the query (GC won the race) — the row is gone
+ // or going; either way the next pass won't see it. A row that
+ // PERSISTS with no bytes is a durability incident the audit
+ // pass reports.
+ continue
+ }
+ return done, err
+ }
+ w, err := m.mirror.create()
+ if err != nil {
+ r.Close()
+ return done, err
+ }
+ hasher := sha256.New()
+ _, copyErr := io.Copy(io.MultiWriter(w, hasher), r)
+ r.Close()
+ if copyErr != nil {
+ w.abort()
+ return done, copyErr
+ }
+ sum := hex.EncodeToString(hasher.Sum(nil))
+ if row.SHA != "" && sum != row.SHA {
+ // The PRIMARY's bytes don't match what was uploaded. Never mirror
+ // corruption — leave the row unmirrored and shout; the mirror may
+ // be about to become the only good copy there is.
+ w.abort()
+ log.Printf("blobs: INTEGRITY: primary bytes for %s hash %.12s, recorded %.12s — not mirrored", row.ID, sum, row.SHA)
+ continue
+ }
+ if err := w.commit(row.ID); err != nil {
+ return done, err
+ }
+ if err := db.markMirrored(row.ID, sum); err != nil {
+ return done, err
+ }
+ done++
+ }
+ return done, nil
+}
+
+// ---- the audit sweep --------------------------------------------------------
+
+// blobAuditResult is one pass's verdict, per store — returned (not just
+// logged) so tests can pin the behavior and Phase B can publish it.
+type blobAuditResult struct {
+ Checked, BadPrimary, BadMirror, MissingPrimary, MissingMirror int
+}
+
+// sampleHashedBlobs picks up to k random mirrored, hashed rows to audit.
+func (db *DB) sampleHashedBlobs(k int) (map[string]string, error) {
+ rows, err := db.sql.Query(
+ `SELECT id, sha256 FROM blobs
+ WHERE sha256 != '' AND mirrored_at > 0 AND length(data) = 0
+ ORDER BY RANDOM() LIMIT ?`, k)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ out := map[string]string{}
+ for rows.Next() {
+ var id, sha string
+ if err := rows.Scan(&id, &sha); err != nil {
+ return nil, err
+ }
+ out[id] = sha
+ }
+ return out, rows.Err()
+}
+
+func hashFromStore(store blobStore, id string) (string, error) {
+ r, _, _, err := store.open(id)
+ if err != nil {
+ return "", err
+ }
+ defer r.Close()
+ hasher := sha256.New()
+ if _, err := io.Copy(hasher, r); err != nil {
+ return "", err
+ }
+ return hex.EncodeToString(hasher.Sum(nil)), nil
+}
+
+// auditOnce full-reads k sampled blobs from EACH store and verifies them
+// against the recorded hash — probabilistic coverage over time at a few
+// requests per pass (docs/federated-blob-storage.md § audit sweep). Mismatches
+// and misses are loud: an integrity failure on either store is the signal the
+// whole trust-but-verify model hangs on.
+func (db *DB) auditBlobsOnce(k int) (blobAuditResult, error) {
+ var res blobAuditResult
+ m, ok := db.blobs.(*mirroredBlobStore)
+ if !ok {
+ return res, nil
+ }
+ sample, err := db.sampleHashedBlobs(k)
+ if err != nil {
+ return res, err
+ }
+ check := func(store blobStore, id, want string, bad, missing *int, name string) {
+ got, err := hashFromStore(store, id)
+ switch {
+ case errors.Is(err, fs.ErrNotExist):
+ *missing++
+ log.Printf("blobs: AUDIT: %s MISSING from %s", id, name)
+ case err != nil:
+ // Transient store trouble isn't an integrity verdict — skip, the
+ // next pass re-samples.
+ case got != want:
+ *bad++
+ log.Printf("blobs: AUDIT: %s CORRUPT on %s (hash %.12s, recorded %.12s)", id, name, got, want)
+ }
+ }
+ for id, want := range sample {
+ res.Checked++
+ check(m.primary, id, want, &res.BadPrimary, &res.MissingPrimary, "primary")
+ check(m.mirror, id, want, &res.BadMirror, &res.MissingMirror, "mirror")
+ }
+ return res, nil
+}
+
+// ---- the loops ---------------------------------------------------------------
+
+// blobMirrorKick nudges the sweep after an upload so a fresh blob mirrors
+// within seconds while the space is awake (a hibernating space's unmirrored
+// tail waits for its next wake — the honest Phase A window).
+func (s *Server) kickBlobMirror() {
+ if s.blobMirrorKick == nil {
+ return
+ }
+ select {
+ case s.blobMirrorKick <- struct{}{}:
+ default: // a kick is already pending — the sweep drains everything anyway
+ }
+}
+
+// startBlobMirror launches the mirror + audit loops when the store is
+// mirrored. Cadences: the mirror ticks every 5m (plus upload kicks) and
+// drains in batches; the audit samples a handful hourly — both idle at zero
+// S3 requests when there's nothing to do (#836's rule).
+func (s *Server) startBlobMirror() {
+ if _, ok := s.db.blobs.(*mirroredBlobStore); !ok {
+ return
+ }
+ s.blobMirrorKick = make(chan struct{}, 1)
+ go func() {
+ tick := time.NewTicker(5 * time.Minute)
+ defer tick.Stop()
+ for {
+ select {
+ case <-tick.C:
+ case <-s.blobMirrorKick:
+ }
+ for {
+ n, err := s.db.mirrorOnce(32)
+ if err != nil {
+ log.Printf("blobs: mirror sweep: %v (retrying next tick)", err)
+ break
+ }
+ if n < 32 {
+ break // drained
+ }
+ }
+ }
+ }()
+ go func() {
+ tick := time.NewTicker(1 * time.Hour)
+ defer tick.Stop()
+ for range tick.C {
+ if res, err := s.db.auditBlobsOnce(3); err != nil {
+ log.Printf("blobs: audit sweep: %v", err)
+ } else if res.BadPrimary+res.BadMirror+res.MissingPrimary+res.MissingMirror > 0 {
+ log.Printf("blobs: AUDIT FAILURES this pass: %+v", res)
+ }
+ }
+ }()
+ log.Printf("blobs: mirror + audit sweeps on")
+}
diff --git a/blobmirror_test.go b/blobmirror_test.go
new file mode 100644
index 0000000..a743620
--- /dev/null
+++ b/blobmirror_test.go
@@ -0,0 +1,236 @@
+package main
+
+import (
+ "bytes"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "io"
+ "io/fs"
+ "net/http"
+ "net/http/httptest"
+ "path/filepath"
+ "testing"
+)
+
+// newMirroredTestDB builds a real DB whose store is primary+mirror over two
+// fake S3 servers — the exact construction openDB performs from env.
+func newMirroredTestDB(t *testing.T) (*DB, *fakeS3, *fakeS3) {
+ t.Helper()
+ primary := &fakeS3{objects: map[string][]byte{}}
+ mirror := &fakeS3{objects: map[string][]byte{}}
+ psrv := httptest.NewServer(http.HandlerFunc(primary.handler))
+ msrv := httptest.NewServer(http.HandlerFunc(mirror.handler))
+ t.Cleanup(psrv.Close)
+ t.Cleanup(msrv.Close)
+
+ t.Setenv("AWS_ACCESS_KEY_ID", "test")
+ t.Setenv("AWS_SECRET_ACCESS_KEY", "test")
+ t.Setenv("AWS_REGION", "auto")
+ t.Setenv("LCHAT_BLOB_S3_BUCKET", "prim")
+ t.Setenv("LCHAT_BLOB_S3_ENDPOINT", psrv.URL)
+ t.Setenv("LCHAT_BLOB_S3_MIRROR_BUCKET", "mirr")
+ t.Setenv("LCHAT_BLOB_S3_MIRROR_ENDPOINT", msrv.URL)
+ // The mirror carries its own key pair (a different provider) — exercises
+ // the static-credentials path.
+ t.Setenv("LCHAT_BLOB_S3_MIRROR_ACCESS_KEY", "mirror-key")
+ t.Setenv("LCHAT_BLOB_S3_MIRROR_SECRET_KEY", "mirror-secret")
+
+ db, err := openDB(filepath.Join(t.TempDir(), "space.example.com.db"))
+ if err != nil {
+ t.Fatalf("openDB: %v", err)
+ }
+ t.Cleanup(func() { db.sql.Close() })
+ if _, ok := db.blobs.(*mirroredBlobStore); !ok {
+ t.Fatalf("store is %T, want mirroredBlobStore", db.blobs)
+ }
+ return db, primary, mirror
+}
+
+// The Phase A loop: a put lands on the primary only; the sweep copies it to
+// the mirror, records mirrored_at, and a second pass is a no-op (idempotence
+// = zero further requests). Reads survive a lost primary object via the
+// mirror, and deletes reap both stores.
+func TestBlobMirrorSweep(t *testing.T) {
+ db, primary, mirror := newMirroredTestDB(t)
+ payload := []byte("ciphertext-bytes-0123456789")
+
+ if err := db.putBlob("blobA", "t1", payload); err != nil {
+ t.Fatalf("putBlob: %v", err)
+ }
+ if got := len(keysOf(mirror)); got != 0 {
+ t.Fatalf("mirror has %d objects before the sweep", got)
+ }
+
+ n, err := db.mirrorOnce(32)
+ if err != nil || n != 1 {
+ t.Fatalf("mirrorOnce = (%d, %v), want (1, nil)", n, err)
+ }
+ mbytes, ok := mirror.objects["/mirr/space.example.com/blobA"]
+ if !ok || !bytes.Equal(mbytes, payload) {
+ t.Fatalf("mirror copy wrong: ok=%v have %v", ok, keysOf(mirror))
+ }
+
+ // Idempotent: nothing pending → zero puts on a second pass.
+ basePuts := mirror.puts
+ if n, err := db.mirrorOnce(32); err != nil || n != 0 {
+ t.Fatalf("second mirrorOnce = (%d, %v), want (0, nil)", n, err)
+ }
+ if mirror.puts != basePuts {
+ t.Fatalf("idle sweep PUT %d times", mirror.puts-basePuts)
+ }
+
+ // Redundancy pays out: primary object lost → the read serves from the
+ // mirror.
+ delete(primary.objects, "/prim/space.example.com/blobA")
+ r, _, _, err := db.blobs.open("blobA")
+ if err != nil {
+ t.Fatalf("open after primary loss: %v", err)
+ }
+ got, _ := io.ReadAll(r)
+ r.Close()
+ if !bytes.Equal(got, payload) {
+ t.Fatalf("mirror served %q", got)
+ }
+
+ // GC reaps both stores.
+ if err := db.deleteBlobs("t1", []string{"blobA"}); err != nil {
+ t.Fatalf("deleteBlobs: %v", err)
+ }
+ if len(keysOf(mirror)) != 0 {
+ t.Fatalf("mirror kept %v after delete", keysOf(mirror))
+ }
+}
+
+// A pre-hash row (sha256 = '') gets its hash backfilled by the sweep — the
+// migration path for every blob uploaded before this feature.
+func TestBlobMirrorBackfillsHash(t *testing.T) {
+ db, _, _ := newMirroredTestDB(t)
+ payload := []byte("old blob, no hash on record")
+ if err := db.putBlob("oldblob", "t1", payload); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := db.sql.Exec("UPDATE blobs SET sha256 = '' WHERE id = 'oldblob'"); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := db.mirrorOnce(32); err != nil {
+ t.Fatal(err)
+ }
+ var sha string
+ if err := db.sql.QueryRow("SELECT sha256 FROM blobs WHERE id = 'oldblob'").Scan(&sha); err != nil {
+ t.Fatal(err)
+ }
+ want := sha256.Sum256(payload)
+ if sha != hex.EncodeToString(want[:]) {
+ t.Fatalf("backfilled hash = %.12s, want %.12s", sha, hex.EncodeToString(want[:]))
+ }
+}
+
+// A primary whose bytes no longer match the recorded hash is NEVER mirrored —
+// corruption must not spread to the copy — and the row stays unmirrored so it
+// keeps being noticed.
+func TestBlobMirrorRefusesCorruptPrimary(t *testing.T) {
+ db, primary, mirror := newMirroredTestDB(t)
+ if err := db.putBlob("blobC", "t1", []byte("genuine bytes")); err != nil {
+ t.Fatal(err)
+ }
+ primary.objects["/prim/space.example.com/blobC"] = []byte("tampered bytes")
+
+ n, err := db.mirrorOnce(32)
+ if err != nil || n != 0 {
+ t.Fatalf("mirrorOnce = (%d, %v), want (0, nil) — corrupt blob must not count as mirrored", n, err)
+ }
+ if len(keysOf(mirror)) != 0 {
+ t.Fatalf("corruption reached the mirror: %v", keysOf(mirror))
+ }
+ var mirroredAt int64
+ if err := db.sql.QueryRow("SELECT mirrored_at FROM blobs WHERE id = 'blobC'").Scan(&mirroredAt); err != nil || mirroredAt != 0 {
+ t.Fatalf("mirrored_at = %d, want 0 (still pending)", mirroredAt)
+ }
+}
+
+// Legacy in-DB rows (bytes in the data column) never enter the sweep — the
+// DB's own replication carries them.
+func TestBlobMirrorSkipsLegacyRows(t *testing.T) {
+ db, _, mirror := newMirroredTestDB(t)
+ if _, err := db.sql.Exec(
+ "INSERT INTO blobs(id, data, size, thread_id, created_at) VALUES('legacy', X'AABB', 2, 't1', 1)"); err != nil {
+ t.Fatal(err)
+ }
+ if n, err := db.mirrorOnce(32); err != nil || n != 0 {
+ t.Fatalf("mirrorOnce = (%d, %v), want (0, nil)", n, err)
+ }
+ if len(keysOf(mirror)) != 0 {
+ t.Fatalf("legacy row reached the mirror: %v", keysOf(mirror))
+ }
+}
+
+// The audit sweep catches a corrupted or missing copy on either store and
+// says which.
+func TestBlobAudit(t *testing.T) {
+ db, primary, mirror := newMirroredTestDB(t)
+ if err := db.putBlob("blobD", "t1", []byte("audited bytes")); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := db.mirrorOnce(32); err != nil {
+ t.Fatal(err)
+ }
+
+ // All healthy: a clean pass.
+ res, err := db.auditBlobsOnce(5)
+ if err != nil || res.Checked != 1 || res.BadPrimary+res.BadMirror+res.MissingPrimary+res.MissingMirror != 0 {
+ t.Fatalf("healthy audit = %+v, %v", res, err)
+ }
+
+ // Corrupt the mirror; lose the primary. Both verdicts land.
+ mirror.objects["/mirr/space.example.com/blobD"] = []byte("rot")
+ delete(primary.objects, "/prim/space.example.com/blobD")
+ res, err = db.auditBlobsOnce(5)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if res.BadMirror != 1 {
+ t.Fatalf("mirror corruption missed: %+v", res)
+ }
+ // The primary miss falls through to the DISK fallback, which still holds
+ // nothing for an S3-era blob → missing.
+ if res.MissingPrimary != 1 {
+ t.Fatalf("primary loss missed: %+v", res)
+ }
+}
+
+// A mirror configured without a primary is a config mistake openDB refuses.
+func TestBlobMirrorRequiresPrimary(t *testing.T) {
+ t.Setenv("LCHAT_BLOB_S3_BUCKET", "")
+ t.Setenv("LCHAT_BLOB_S3_MIRROR_BUCKET", "mirr")
+ if _, err := openDB(filepath.Join(t.TempDir(), "x.db")); err == nil {
+ t.Fatal("openDB accepted a mirror without a primary")
+ }
+}
+
+// The upload handler records the streamed sha256 — putBlob's twin assertion
+// (the handler itself is exercised in handlers tests; this pins the hash).
+func TestPutBlobRecordsHash(t *testing.T) {
+ db, _, _ := newMirroredTestDB(t)
+ payload := []byte("hash me")
+ if err := db.putBlob("hashed", "", payload); err != nil {
+ t.Fatal(err)
+ }
+ var sha string
+ if err := db.sql.QueryRow("SELECT sha256 FROM blobs WHERE id = 'hashed'").Scan(&sha); err != nil {
+ t.Fatal(err)
+ }
+ want := sha256.Sum256(payload)
+ if sha != hex.EncodeToString(want[:]) {
+ t.Fatalf("recorded hash = %.12s, want %.12s", sha, hex.EncodeToString(want[:]))
+ }
+}
+
+// The fs.ErrNotExist contract survives composition: an id nobody holds errors
+// identically through the mirrored store.
+func TestMirroredStoreNotExist(t *testing.T) {
+ db, _, _ := newMirroredTestDB(t)
+ if _, _, _, err := db.blobs.open("ghost"); !errors.Is(err, fs.ErrNotExist) {
+ t.Fatalf("open(ghost) = %v, want fs.ErrNotExist", err)
+ }
+}
diff --git a/db.go b/db.go
index 3a88124..9f94726 100644
--- a/db.go
+++ b/db.go
@@ -288,16 +288,30 @@ func openDB(path string) (*DB, error) {
// The blobStore seam's switch (#550 Phase 1): LCHAT_BLOB_S3_BUCKET set
// puts NEW attachment bytes in the object store, with the disk dir as the
// fallback for pre-flip blobs and the staging spool. Unset = today's
- // on-disk layout, byte-identical behavior.
+ // on-disk layout, byte-identical behavior. LCHAT_BLOB_S3_MIRROR_* beside
+ // it adds the second copy on a second provider (federated blobs Phase A,
+ // #992 — the mirror sweep in blobmirror.go keeps it converged).
var blobs blobStore = disk
if cfg := s3BlobConfigFromEnv(); cfg != nil {
host := strings.TrimSuffix(filepath.Base(path), filepath.Ext(path))
- s3s, err := newS3BlobStore(cfg, disk, host)
+ s3s, err := newS3BlobStore(cfg, disk.dir, disk, host)
if err != nil {
return nil, err
}
blobs = s3s
log.Printf("blobs: s3 store on (bucket %s, prefix %s)", cfg.Bucket, s3s.prefix)
+ if mcfg := s3BlobMirrorConfigFromEnv(); mcfg != nil {
+ mirror, err := newS3BlobStore(mcfg, disk.dir, nil, host)
+ if err != nil {
+ return nil, err
+ }
+ blobs = &mirroredBlobStore{primary: s3s, mirror: mirror}
+ log.Printf("blobs: mirror store on (bucket %s, prefix %s)", mcfg.Bucket, mirror.prefix)
+ }
+ } else if s3BlobMirrorConfigFromEnv() != nil {
+ // A mirror with no primary is a config mistake, not a mode — the
+ // disk layout's second copy is the fleet backup's job.
+ return nil, errors.New("LCHAT_BLOB_S3_MIRROR_* set without LCHAT_BLOB_S3_BUCKET")
}
db := &DB{sql: d, blobs: blobs,
vapidPath: strings.TrimSuffix(path, filepath.Ext(path)) + ".vapid"}
@@ -769,6 +783,16 @@ CREATE UNIQUE INDEX IF NOT EXISTS idx_checkpoints_seq_from ON checkpoints(seq_fr
// Blobs gain a thread tag (for thread-scoped GC + quota accounting). Older
// blobs (branding, pre-disk photos) stay NULL and are never thread-GC'd.
"ALTER TABLE blobs ADD COLUMN thread_id TEXT",
+ // Federated blobs Phase A (#992): the ciphertext's sha256 (hex),
+ // recorded at upload — the integrity anchor mirror and audit sweeps
+ // verify against. '' = pre-hash row; the mirror sweep backfills it
+ // (it reads the full bytes anyway).
+ "ALTER TABLE blobs ADD COLUMN sha256 TEXT NOT NULL DEFAULT ''",
+ // When the mirror store last received this blob (unix seconds; 0 =
+ // unmirrored — what the mirror sweep selects on). Reset to 0 fleet-wide
+ // when the mirror TARGET changes (documented beside the env vars);
+ // Phase B's store registry replaces this with per-store placement.
+ "ALTER TABLE blobs ADD COLUMN mirrored_at INTEGER NOT NULL DEFAULT 0",
// Roles (docs/conversational-onboarding.md): 'admin' | 'moderator' |
// 'member'. Authorisation only — a role grants no key material. The admin
// bearer token remains as break-glass; role-admins reach the same admin API
@@ -3987,14 +4011,14 @@ func (db *DB) messagesAfter(threadID string, after int64, limit int) ([]Message,
// the row exists only for quota accounting and thread-scoped GC. threadID ties
// the blob to its thread (empty → NULL, e.g. profile-avatar uploads that aren't
// thread-scoped and are replaced rather than message-GC'd).
-func (db *DB) recordBlob(id, threadID string, size int64) error {
+func (db *DB) recordBlob(id, threadID string, size int64, sha256Hex string) error {
var tid any
if threadID != "" {
tid = threadID
}
_, err := db.sql.Exec(
- "INSERT INTO blobs(id, data, size, thread_id, created_at) VALUES(?, X'', ?, ?, ?)",
- id, size, tid, time.Now().Unix(),
+ "INSERT INTO blobs(id, data, size, thread_id, created_at, sha256) VALUES(?, X'', ?, ?, ?, ?)",
+ id, size, tid, time.Now().Unix(), sha256Hex,
)
return err
}
@@ -4014,7 +4038,8 @@ func (db *DB) putBlob(id, threadID string, data []byte) error {
if err := w.commit(id); err != nil {
return err
}
- if err := db.recordBlob(id, threadID, int64(len(data))); err != nil {
+ sum := sha256.Sum256(data)
+ if err := db.recordBlob(id, threadID, int64(len(data)), hex.EncodeToString(sum[:])); err != nil {
_ = db.blobs.remove(id)
return err
}
diff --git a/docs/planet-scale-guide/reference.html b/docs/planet-scale-guide/reference.html
index 6ca4843..e36d1ac 100644
--- a/docs/planet-scale-guide/reference.html
+++ b/docs/planet-scale-guide/reference.html
@@ -109,7 +109,8 @@
<tr><td><code>LCHAT_ADMIN_TOKEN</code></td><td>The instance&rsquo;s admin token. The database stores only its hash; <strong>this file is the only recoverable copy</strong>.</td></tr>
<tr><td><code>LCHAT_APP_STORE_URL</code>, <code>LCHAT_HOME_ORIGIN</code>, <code>LCHAT_PUSH_RELAY_URL</code>, <code>LCHAT_PUSH_RELAY_TOKEN</code></td><td>Per-instance pointers to the app-store listing, the home server, and the push relay (with its per-instance bearer token).</td></tr>
<tr><td><code>LCHAT_BLOB_QUOTA_BYTES</code></td><td>Attachment-storage cap for the space (written by the control daemon&rsquo;s quota endpoint; uploads beyond it are refused).</td></tr>
- <tr><td><code>LCHAT_BLOB_S3_BUCKET</code> (+ <code>_PREFIX</code>, <code>_REGION</code>, <code>_ENDPOINT</code>)</td><td>Attachment bytes in an S3-compatible object store (R2 intended — zero egress). Set = new uploads land there, the on-disk dir keeps serving pre-flip blobs; unset = today&rsquo;s disk layout. Prefix defaults to the host. Credentials via <code>AWS_ACCESS_KEY_ID</code>/<code>_SECRET_ACCESS_KEY</code> in the same file. <span class="status partial">built, not yet flipped</span></td></tr>
+ <tr><td><code>LCHAT_BLOB_S3_BUCKET</code> (+ <code>_PREFIX</code>, <code>_REGION</code>, <code>_ENDPOINT</code>, <code>_ACCESS_KEY</code>, <code>_SECRET_KEY</code>)</td><td>Attachment bytes in an S3-compatible object store. Set = new uploads land there, the on-disk dir keeps serving pre-flip blobs; unset = today&rsquo;s disk layout. Prefix defaults to the host; key pair falls back to the AWS default chain. <span class="status partial">built, not yet flipped</span></td></tr>
+ <tr><td><code>LCHAT_BLOB_S3_MIRROR_*</code> (same six)</td><td>A second copy on a second provider (federated blobs Phase A — <code>docs/federated-blob-storage.md</code>): an idempotent sweep mirrors every store-backed blob, reads chain primary&rarr;mirror&rarr;disk, deletes reap both, and an hourly audit spot-checks both stores against each blob&rsquo;s recorded sha256. Requires the primary. Changing the mirror <em>target</em> later: <code>UPDATE blobs SET mirrored_at=0</code> re-mirrors everything. <span class="status partial">built, not yet flipped</span></td></tr>
</table>
</div>
diff --git a/main.go b/main.go
index da6c2af..4e3a407 100644
--- a/main.go
+++ b/main.go
@@ -8,6 +8,7 @@ import (
"database/sql"
"embed"
"encoding/base64"
+ "encoding/hex"
"encoding/json"
"errors"
"flag"
@@ -92,6 +93,9 @@ type Server struct {
apns apnsSender // nil unless APNs (direct p8 or relay) is configured
maxUpload int64 // per-upload attachment cap (LCHAT_MAX_UPLOAD_BYTES)
blobQuota int64 // per-instance total blob-bytes ceiling (LCHAT_BLOB_QUOTA_BYTES)
+ // blobMirrorKick nudges the blob mirror sweep after an upload
+ // (blobmirror.go, #992). Nil unless the store is mirrored.
+ blobMirrorKick chan struct{}
// The usage-pipe relay (usage.go, #857): where sidecar usage reports go
// and the per-space meter token that authenticates them. Pushed into the
// env by the provisioner; empty = unmetered (self-hosted / BYO key).
@@ -316,6 +320,9 @@ func main() {
go s.runUpdateChecker()
// Presence idle sweep (presence.go). Tests call sweepPresence directly.
go s.runPresenceSweeper()
+ // Blob mirror + audit sweeps (blobmirror.go, #992). No-op unless the
+ // store is mirrored; tests call mirrorOnce/auditBlobsOnce directly.
+ s.startBlobMirror()
// Tamper evidence: fold new journal rows into signed checkpoints
// (checkpoint.go). Tests call sweepCheckpoint directly.
go s.runCheckpointer()
@@ -2905,13 +2912,15 @@ func (s *Server) uploadBlob(w http.ResponseWriter, r *http.Request) {
}
// Stream into a staged store write (never buffer a whole video in RAM);
// commit under the id only once it's fully written, so a partial/aborted
- // upload is never served.
+ // upload is never served. The sha256 rides the same pass — it's the
+ // integrity anchor the blob mirror/audit sweeps verify against (#992).
blob, err := s.db.blobs.create()
if err != nil {
httpError(w, http.StatusInternalServerError, "storage error")
return
}
- size, err := io.Copy(blob, r.Body)
+ hasher := sha256.New()
+ size, err := io.Copy(io.MultiWriter(blob, hasher), r.Body)
if err != nil {
blob.abort()
// MaxBytesReader trips here when the body exceeds the cap.
@@ -2936,11 +2945,12 @@ func (s *Server) uploadBlob(w http.ResponseWriter, r *http.Request) {
// Tag with the (server-known, non-secret) thread id so the blob can be GC'd
// with its thread / when the referencing message is deleted. Optional: avatar
// uploads carry no thread.
- if err := s.db.recordBlob(id, r.URL.Query().Get("thread"), size); err != nil {
+ if err := s.db.recordBlob(id, r.URL.Query().Get("thread"), size, hex.EncodeToString(hasher.Sum(nil))); err != nil {
_ = s.db.blobs.remove(id)
httpError(w, http.StatusInternalServerError, "db error")
return
}
+ s.kickBlobMirror() // a fresh blob mirrors promptly while the space is awake
writeJSON(w, http.StatusOK, map[string]any{"id": id})
}
diff --git a/s3blobstore.go b/s3blobstore.go
index a464754..d63f694 100644
--- a/s3blobstore.go
+++ b/s3blobstore.go
@@ -58,6 +58,7 @@ import (
"github.com/aws/aws-sdk-go-v2/aws"
awsconfig "github.com/aws/aws-sdk-go-v2/config"
+ "github.com/aws/aws-sdk-go-v2/credentials"
awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
"github.com/aws/smithy-go"
)
@@ -75,8 +76,8 @@ type s3BlobStore struct {
client *awss3.Client
bucket string
prefix string
- // fallback serves blobs uploaded before the S3 flip (and absorbs removes
- // for them). Always the instance's disk store today.
+ // fallback serves blobs the remote lacks (pre-flip disk blobs for the
+ // primary store). Nil for a mirror store — a miss there is just a miss.
fallback blobStore
// spoolDir stages in-flight uploads — the disk store's dir, so the
// temp-name convention ("." never appears in a valid blob id) keeps
@@ -90,28 +91,45 @@ type s3BlobConfig struct {
Prefix string
Region string
Endpoint string
+ // Static credentials for THIS store — primary and mirror are different
+ // providers with different key pairs, and the AWS default chain can only
+ // hold one. Empty = the default chain (the primary's usual shape).
+ AccessKey string
+ SecretKey string
}
-func s3BlobConfigFromEnv() *s3BlobConfig {
- bucket := strings.TrimSpace(os.Getenv("LCHAT_BLOB_S3_BUCKET"))
+// s3BlobEnv reads one store's env quartet (+ key pair) under a prefix:
+// LCHAT_BLOB_S3_* for the primary, LCHAT_BLOB_S3_MIRROR_* for the mirror.
+func s3BlobEnv(envPrefix string) *s3BlobConfig {
+ bucket := strings.TrimSpace(os.Getenv(envPrefix + "BUCKET"))
if bucket == "" {
return nil
}
return &s3BlobConfig{
- Bucket: bucket,
- Prefix: strings.Trim(strings.TrimSpace(os.Getenv("LCHAT_BLOB_S3_PREFIX")), "/"),
- Region: strings.TrimSpace(os.Getenv("LCHAT_BLOB_S3_REGION")),
- Endpoint: strings.TrimSpace(os.Getenv("LCHAT_BLOB_S3_ENDPOINT")),
+ Bucket: bucket,
+ Prefix: strings.Trim(strings.TrimSpace(os.Getenv(envPrefix+"PREFIX")), "/"),
+ Region: strings.TrimSpace(os.Getenv(envPrefix + "REGION")),
+ Endpoint: strings.TrimSpace(os.Getenv(envPrefix + "ENDPOINT")),
+ AccessKey: strings.TrimSpace(os.Getenv(envPrefix + "ACCESS_KEY")),
+ SecretKey: strings.TrimSpace(os.Getenv(envPrefix + "SECRET_KEY")),
}
}
-// newS3BlobStore builds the store over an existing disk store (the fallback +
-// spool home). defaultPrefix is the DB basename — used when the env names none.
-func newS3BlobStore(cfg *s3BlobConfig, disk *diskBlobStore, defaultPrefix string) (*s3BlobStore, error) {
+func s3BlobConfigFromEnv() *s3BlobConfig { return s3BlobEnv("LCHAT_BLOB_S3_") }
+func s3BlobMirrorConfigFromEnv() *s3BlobConfig { return s3BlobEnv("LCHAT_BLOB_S3_MIRROR_") }
+
+// newS3BlobStore builds one remote store. fallback may be nil (a mirror);
+// spoolDir stages writes; defaultPrefix is the DB basename — used when the
+// env names none.
+func newS3BlobStore(cfg *s3BlobConfig, spoolDir string, fallback blobStore, defaultPrefix string) (*s3BlobStore, error) {
var opts []func(*awsconfig.LoadOptions) error
if cfg.Region != "" {
opts = append(opts, awsconfig.WithRegion(cfg.Region))
}
+ if cfg.AccessKey != "" {
+ opts = append(opts, awsconfig.WithCredentialsProvider(
+ credentials.NewStaticCredentialsProvider(cfg.AccessKey, cfg.SecretKey, "")))
+ }
awsCfg, err := awsconfig.LoadDefaultConfig(context.Background(), opts...)
if err != nil {
return nil, fmt.Errorf("blob s3: load aws config: %w", err)
@@ -133,7 +151,7 @@ func newS3BlobStore(cfg *s3BlobConfig, disk *diskBlobStore, defaultPrefix string
prefix = defaultPrefix
}
return &s3BlobStore{client: client, bucket: cfg.Bucket, prefix: prefix,
- fallback: disk, spoolDir: disk.dir}, nil
+ fallback: fallback, spoolDir: spoolDir}, nil
}
func (s *s3BlobStore) key(id string) string { return path.Join(s.prefix, id) }
@@ -166,9 +184,12 @@ func (s *s3BlobStore) open(id string) (io.ReadSeekCloser, int64, time.Time, erro
head, err := s.client.HeadObject(ctx, &awss3.HeadObjectInput{Bucket: &s.bucket, Key: &key})
if err != nil {
if s3NotFound(err) {
- // Pre-flip blob (or genuinely absent — the fallback answers with
- // fs.ErrNotExist either way, which is the contract).
- return s.fallback.open(id)
+ if s.fallback != nil {
+ // Pre-flip blob (or genuinely absent — the fallback answers
+ // with fs.ErrNotExist either way, which is the contract).
+ return s.fallback.open(id)
+ }
+ return nil, 0, time.Time{}, fs.ErrNotExist
}
return nil, 0, time.Time{}, err
}
@@ -192,7 +213,10 @@ func (s *s3BlobStore) remove(id string) error {
}
// A pre-flip blob's bytes live on disk — GC must reap those too. The
// fallback treats a missing file as success, so this is free for S3-era ids.
- return s.fallback.remove(id)
+ if s.fallback != nil {
+ return s.fallback.remove(id)
+ }
+ return nil
}
type s3BlobWriter struct {
diff --git a/s3blobstore_test.go b/s3blobstore_test.go
index 7cfcda0..5b5c463 100644
--- a/s3blobstore_test.go
+++ b/s3blobstore_test.go
@@ -95,7 +95,7 @@ func newTestS3Store(t *testing.T) (*s3BlobStore, *fakeS3, *diskBlobStore) {
store, err := newS3BlobStore(&s3BlobConfig{
Bucket: "test-bucket",
Endpoint: srv.URL,
- }, disk, "space.example.com")
+ }, disk.dir, disk, "space.example.com")
if err != nil {
t.Fatal(err)
}