| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "crypto/sha256" |
| 6 | "encoding/hex" |
| 7 | "errors" |
| 8 | "io" |
| 9 | "io/fs" |
| 10 | "net/http" |
| 11 | "net/http/httptest" |
| 12 | "path/filepath" |
| 13 | "testing" |
| 14 | ) |
| 15 | |
| 16 | // newMirroredTestDB builds a real DB whose store is primary+mirror over two |
| 17 | // fake S3 servers — the exact construction openDB performs from env. |
| 18 | func newMirroredTestDB(t *testing.T) (*DB, *fakeS3, *fakeS3) { |
| 19 | t.Helper() |
| 20 | primary := &fakeS3{objects: map[string][]byte{}} |
| 21 | mirror := &fakeS3{objects: map[string][]byte{}} |
| 22 | psrv := httptest.NewServer(http.HandlerFunc(primary.handler)) |
| 23 | msrv := httptest.NewServer(http.HandlerFunc(mirror.handler)) |
| 24 | t.Cleanup(psrv.Close) |
| 25 | t.Cleanup(msrv.Close) |
| 26 | |
| 27 | t.Setenv("AWS_ACCESS_KEY_ID", "test") |
| 28 | t.Setenv("AWS_SECRET_ACCESS_KEY", "test") |
| 29 | t.Setenv("AWS_REGION", "auto") |
| 30 | t.Setenv("LCHAT_BLOB_S3_BUCKET", "prim") |
| 31 | t.Setenv("LCHAT_BLOB_S3_ENDPOINT", psrv.URL) |
| 32 | t.Setenv("LCHAT_BLOB_S3_MIRROR_BUCKET", "mirr") |
| 33 | t.Setenv("LCHAT_BLOB_S3_MIRROR_ENDPOINT", msrv.URL) |
| 34 | // The mirror carries its own key pair (a different provider) — exercises |
| 35 | // the static-credentials path. |
| 36 | t.Setenv("LCHAT_BLOB_S3_MIRROR_ACCESS_KEY", "mirror-key") |
| 37 | t.Setenv("LCHAT_BLOB_S3_MIRROR_SECRET_KEY", "mirror-secret") |
| 38 | |
| 39 | db, err := openDB(filepath.Join(t.TempDir(), "space.example.com.db")) |
| 40 | if err != nil { |
| 41 | t.Fatalf("openDB: %v", err) |
| 42 | } |
| 43 | t.Cleanup(func() { db.sql.Close() }) |
| 44 | if _, ok := db.blobs.(*mirroredBlobStore); !ok { |
| 45 | t.Fatalf("store is %T, want mirroredBlobStore", db.blobs) |
| 46 | } |
| 47 | return db, primary, mirror |
| 48 | } |
| 49 | |
| 50 | // The Phase A loop: a put lands on the primary only; the sweep copies it to |
| 51 | // the mirror, records mirrored_at, and a second pass is a no-op (idempotence |
| 52 | // = zero further requests). Reads survive a lost primary object via the |
| 53 | // mirror, and deletes reap both stores. |
| 54 | func TestBlobMirrorSweep(t *testing.T) { |
| 55 | db, primary, mirror := newMirroredTestDB(t) |
| 56 | payload := []byte("ciphertext-bytes-0123456789") |
| 57 | |
| 58 | if err := db.putBlob("blobA", "t1", payload); err != nil { |
| 59 | t.Fatalf("putBlob: %v", err) |
| 60 | } |
| 61 | if got := len(keysOf(mirror)); got != 0 { |
| 62 | t.Fatalf("mirror has %d objects before the sweep", got) |
| 63 | } |
| 64 | |
| 65 | n, err := db.mirrorOnce(32) |
| 66 | if err != nil || n != 1 { |
| 67 | t.Fatalf("mirrorOnce = (%d, %v), want (1, nil)", n, err) |
| 68 | } |
| 69 | mbytes, ok := mirror.objects["/mirr/space.example.com/blobA"] |
| 70 | if !ok || !bytes.Equal(mbytes, payload) { |
| 71 | t.Fatalf("mirror copy wrong: ok=%v have %v", ok, keysOf(mirror)) |
| 72 | } |
| 73 | |
| 74 | // Idempotent: nothing pending → zero puts on a second pass. |
| 75 | basePuts := mirror.puts |
| 76 | if n, err := db.mirrorOnce(32); err != nil || n != 0 { |
| 77 | t.Fatalf("second mirrorOnce = (%d, %v), want (0, nil)", n, err) |
| 78 | } |
| 79 | if mirror.puts != basePuts { |
| 80 | t.Fatalf("idle sweep PUT %d times", mirror.puts-basePuts) |
| 81 | } |
| 82 | |
| 83 | // Redundancy pays out: primary object lost → the read serves from the |
| 84 | // mirror. |
| 85 | delete(primary.objects, "/prim/space.example.com/blobA") |
| 86 | r, _, _, err := db.blobs.open("blobA") |
| 87 | if err != nil { |
| 88 | t.Fatalf("open after primary loss: %v", err) |
| 89 | } |
| 90 | got, _ := io.ReadAll(r) |
| 91 | r.Close() |
| 92 | if !bytes.Equal(got, payload) { |
| 93 | t.Fatalf("mirror served %q", got) |
| 94 | } |
| 95 | |
| 96 | // GC reaps both stores. |
| 97 | if err := db.deleteBlobs("t1", []string{"blobA"}); err != nil { |
| 98 | t.Fatalf("deleteBlobs: %v", err) |
| 99 | } |
| 100 | if len(keysOf(mirror)) != 0 { |
| 101 | t.Fatalf("mirror kept %v after delete", keysOf(mirror)) |
| 102 | } |
| 103 | } |
| 104 | |
| 105 | // A pre-hash row (sha256 = ”) gets its hash backfilled by the sweep — the |
| 106 | // migration path for every blob uploaded before this feature. |
| 107 | func TestBlobMirrorBackfillsHash(t *testing.T) { |
| 108 | db, _, _ := newMirroredTestDB(t) |
| 109 | payload := []byte("old blob, no hash on record") |
| 110 | if err := db.putBlob("oldblob", "t1", payload); err != nil { |
| 111 | t.Fatal(err) |
| 112 | } |
| 113 | if _, err := db.sql.Exec("UPDATE blobs SET sha256 = '' WHERE id = 'oldblob'"); err != nil { |
| 114 | t.Fatal(err) |
| 115 | } |
| 116 | if _, err := db.mirrorOnce(32); err != nil { |
| 117 | t.Fatal(err) |
| 118 | } |
| 119 | var sha string |
| 120 | if err := db.sql.QueryRow("SELECT sha256 FROM blobs WHERE id = 'oldblob'").Scan(&sha); err != nil { |
| 121 | t.Fatal(err) |
| 122 | } |
| 123 | want := sha256.Sum256(payload) |
| 124 | if sha != hex.EncodeToString(want[:]) { |
| 125 | t.Fatalf("backfilled hash = %.12s, want %.12s", sha, hex.EncodeToString(want[:])) |
| 126 | } |
| 127 | } |
| 128 | |
| 129 | // A primary whose bytes no longer match the recorded hash is NEVER mirrored — |
| 130 | // corruption must not spread to the copy — and the row stays unmirrored so it |
| 131 | // keeps being noticed. |
| 132 | func TestBlobMirrorRefusesCorruptPrimary(t *testing.T) { |
| 133 | db, primary, mirror := newMirroredTestDB(t) |
| 134 | if err := db.putBlob("blobC", "t1", []byte("genuine bytes")); err != nil { |
| 135 | t.Fatal(err) |
| 136 | } |
| 137 | primary.objects["/prim/space.example.com/blobC"] = []byte("tampered bytes") |
| 138 | |
| 139 | n, err := db.mirrorOnce(32) |
| 140 | if err != nil || n != 0 { |
| 141 | t.Fatalf("mirrorOnce = (%d, %v), want (0, nil) — corrupt blob must not count as mirrored", n, err) |
| 142 | } |
| 143 | if len(keysOf(mirror)) != 0 { |
| 144 | t.Fatalf("corruption reached the mirror: %v", keysOf(mirror)) |
| 145 | } |
| 146 | var mirroredAt int64 |
| 147 | if err := db.sql.QueryRow("SELECT mirrored_at FROM blobs WHERE id = 'blobC'").Scan(&mirroredAt); err != nil || mirroredAt != 0 { |
| 148 | t.Fatalf("mirrored_at = %d, want 0 (still pending)", mirroredAt) |
| 149 | } |
| 150 | } |
| 151 | |
| 152 | // Legacy in-DB rows (bytes in the data column) never enter the sweep — the |
| 153 | // DB's own replication carries them. |
| 154 | func TestBlobMirrorSkipsLegacyRows(t *testing.T) { |
| 155 | db, _, mirror := newMirroredTestDB(t) |
| 156 | if _, err := db.sql.Exec( |
| 157 | "INSERT INTO blobs(id, data, size, thread_id, created_at) VALUES('legacy', X'AABB', 2, 't1', 1)"); err != nil { |
| 158 | t.Fatal(err) |
| 159 | } |
| 160 | if n, err := db.mirrorOnce(32); err != nil || n != 0 { |
| 161 | t.Fatalf("mirrorOnce = (%d, %v), want (0, nil)", n, err) |
| 162 | } |
| 163 | if len(keysOf(mirror)) != 0 { |
| 164 | t.Fatalf("legacy row reached the mirror: %v", keysOf(mirror)) |
| 165 | } |
| 166 | } |
| 167 | |
| 168 | // The audit sweep catches a corrupted or missing copy on either store and |
| 169 | // says which. |
| 170 | func TestBlobAudit(t *testing.T) { |
| 171 | db, primary, mirror := newMirroredTestDB(t) |
| 172 | if err := db.putBlob("blobD", "t1", []byte("audited bytes")); err != nil { |
| 173 | t.Fatal(err) |
| 174 | } |
| 175 | if _, err := db.mirrorOnce(32); err != nil { |
| 176 | t.Fatal(err) |
| 177 | } |
| 178 | |
| 179 | // All healthy: a clean pass. |
| 180 | res, err := db.auditBlobsOnce(5) |
| 181 | if err != nil || res.Checked != 1 || res.BadPrimary+res.BadMirror+res.MissingPrimary+res.MissingMirror != 0 { |
| 182 | t.Fatalf("healthy audit = %+v, %v", res, err) |
| 183 | } |
| 184 | |
| 185 | // Corrupt the mirror; lose the primary. Both verdicts land. |
| 186 | mirror.objects["/mirr/space.example.com/blobD"] = []byte("rot") |
| 187 | delete(primary.objects, "/prim/space.example.com/blobD") |
| 188 | res, err = db.auditBlobsOnce(5) |
| 189 | if err != nil { |
| 190 | t.Fatal(err) |
| 191 | } |
| 192 | if res.BadMirror != 1 { |
| 193 | t.Fatalf("mirror corruption missed: %+v", res) |
| 194 | } |
| 195 | // The primary miss falls through to the DISK fallback, which still holds |
| 196 | // nothing for an S3-era blob → missing. |
| 197 | if res.MissingPrimary != 1 { |
| 198 | t.Fatalf("primary loss missed: %+v", res) |
| 199 | } |
| 200 | } |
| 201 | |
| 202 | // A mirror configured without a primary is a config mistake openDB refuses. |
| 203 | func TestBlobMirrorRequiresPrimary(t *testing.T) { |
| 204 | t.Setenv("LCHAT_BLOB_S3_BUCKET", "") |
| 205 | t.Setenv("LCHAT_BLOB_S3_MIRROR_BUCKET", "mirr") |
| 206 | if _, err := openDB(filepath.Join(t.TempDir(), "x.db")); err == nil { |
| 207 | t.Fatal("openDB accepted a mirror without a primary") |
| 208 | } |
| 209 | } |
| 210 | |
| 211 | // The upload handler records the streamed sha256 — putBlob's twin assertion |
| 212 | // (the handler itself is exercised in handlers tests; this pins the hash). |
| 213 | func TestPutBlobRecordsHash(t *testing.T) { |
| 214 | db, _, _ := newMirroredTestDB(t) |
| 215 | payload := []byte("hash me") |
| 216 | if err := db.putBlob("hashed", "", payload); err != nil { |
| 217 | t.Fatal(err) |
| 218 | } |
| 219 | var sha string |
| 220 | if err := db.sql.QueryRow("SELECT sha256 FROM blobs WHERE id = 'hashed'").Scan(&sha); err != nil { |
| 221 | t.Fatal(err) |
| 222 | } |
| 223 | want := sha256.Sum256(payload) |
| 224 | if sha != hex.EncodeToString(want[:]) { |
| 225 | t.Fatalf("recorded hash = %.12s, want %.12s", sha, hex.EncodeToString(want[:])) |
| 226 | } |
| 227 | } |
| 228 | |
| 229 | // The fs.ErrNotExist contract survives composition: an id nobody holds errors |
| 230 | // identically through the mirrored store. |
| 231 | func TestMirroredStoreNotExist(t *testing.T) { |
| 232 | db, _, _ := newMirroredTestDB(t) |
| 233 | if _, _, _, err := db.blobs.open("ghost"); !errors.Is(err, fs.ErrNotExist) { |
| 234 | t.Fatalf("open(ghost) = %v, want fs.ErrNotExist", err) |
| 235 | } |
| 236 | } |
| 237 | |
amadan