amadan

eleven / ocho Public

server: S3-compatible blob store — attachment bytes in R2, behind the #636 seam (#550 Phase 1) (#987)

blobStore implementation #2, per instance via LCHAT_BLOB_S3_BUCKET (unset =
today's disk layout, byte-identical). Wraps the disk store rather than
replacing it: new uploads PUT to the object store, opens/removes consult S3
then fall back to disk, so pre-flip blobs keep serving and the disk set only
shrinks — the Phase 2 migration sweep drains it, marker-free.

Staged writes spool to the blobs dir and upload on commit (a completed PUT is
atomic, so invisible-until-commit holds with no staging keys to GC; the spool
gives R2 the ContentLength it requires). Reads are chunked ranged GETs behind
an io.ReadSeekCloser — http.ServeContent's size probe costs zero requests, a
video scrub costs one 512KB window, sequential serving ~2 requests/MB (#836's
lesson: the request count is the bill). Checksum calculation set to
when-required — the SDK's default aws-chunked framing is what R2 rejects.

Same E2EE posture as disk: ciphertext bytes under server-minted ids; the
object store learns nothing the local disk didn't know. Quota accounting is
untouched (the blobs TABLE was always the source of truth).

Tests: the disk contract suite against the S3 store via a counting fake
(round-trip, abort, staged invisibility, fallback serve+reap), plus the
request-economy pins (0-GET size probe, 1-GET scrub, 2-GET sequential read).
Live flip needs an R2 bucket + per-instance credentials — deliberately not
part of this PR.
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 0b02a55d9e33c09c5a58f1bc3b3c0a4f06d10566 parent a98c3da
5 files changed, +629 −2
  • db.go +15 −1
  • docs/planet-scale-guide/reference.html +1 −0
  • go.mod +1 −1
  • s3blobstore.go +307 −0
  • s3blobstore_test.go +305 −0
diff --git a/db.go b/db.go
index 97c5887..b33c1f7 100644
--- a/db.go
+++ b/db.go
@@ -268,10 +268,24 @@ func openDB(path string) (*DB, error) {
return nil, err
}
}
- blobs, err := newDiskBlobStore(strings.TrimSuffix(path, filepath.Ext(path)) + "-blobs")
+ disk, err := newDiskBlobStore(strings.TrimSuffix(path, filepath.Ext(path)) + "-blobs")
if err != nil {
return nil, err
}
+ // 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.
+ var blobs blobStore = disk
+ if cfg := s3BlobConfigFromEnv(); cfg != nil {
+ host := strings.TrimSuffix(filepath.Base(path), filepath.Ext(path))
+ s3s, err := newS3BlobStore(cfg, disk, host)
+ if err != nil {
+ return nil, err
+ }
+ blobs = s3s
+ log.Printf("blobs: s3 store on (bucket %s, prefix %s)", cfg.Bucket, s3s.prefix)
+ }
db := &DB{sql: d, blobs: blobs,
vapidPath: strings.TrimSuffix(path, filepath.Ext(path)) + ".vapid"}
if err := db.migrate(); err != nil {
diff --git a/docs/planet-scale-guide/reference.html b/docs/planet-scale-guide/reference.html
index 2b895f0..6ca4843 100644
--- a/docs/planet-scale-guide/reference.html
+++ b/docs/planet-scale-guide/reference.html
@@ -109,6 +109,7 @@
<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>
</table>
</div>
diff --git a/go.mod b/go.mod
index 8f60392..5f16f41 100644
--- a/go.mod
+++ b/go.mod
@@ -9,6 +9,7 @@ require (
github.com/aws/aws-sdk-go-v2 v1.42.1
github.com/aws/aws-sdk-go-v2/config v1.32.6
github.com/aws/aws-sdk-go-v2/service/s3 v1.105.0
+ github.com/aws/smithy-go v1.27.3
github.com/benbjohnson/litestream v0.5.14
github.com/coder/websocket v1.8.12
github.com/go-webauthn/webauthn v0.11.2
@@ -34,7 +35,6 @@ require (
github.com/aws/aws-sdk-go-v2/service/sso v1.30.8 // indirect
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12 // indirect
github.com/aws/aws-sdk-go-v2/service/sts v1.41.5 // indirect
- github.com/aws/smithy-go v1.27.3 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/dustin/go-humanize v1.0.1 // indirect
diff --git a/s3blobstore.go b/s3blobstore.go
new file mode 100644
index 0000000..a464754
--- /dev/null
+++ b/s3blobstore.go
@@ -0,0 +1,307 @@
+package main
+
+// 🤖 s3BlobStore — blobStore implementation #2 (#550 Phase 1): attachment
+// bytes in an S3-compatible object store (Cloudflare R2 is the intended
+// target — zero egress; see docs/tier-economics.md). Enabled per instance by
+// LCHAT_BLOB_S3_BUCKET in its env; unset keeps today's on-disk layout, so
+// self-hosters and the fleet flip independently, instance by instance.
+//
+// The store carries the same E2EE posture as disk: bytes are client-side
+// ciphertext, keys/filenames/types never reach here, and object keys are the
+// server-minted blob ids — the object store learns exactly what the local
+// disk knew, nothing more.
+//
+// Shape decisions, in the order they matter:
+//
+// - WRAPS the disk store as a fallback, never replaces it: blobs uploaded
+// before the flip live on disk, and open/remove consult S3 first, then
+// disk. New writes always land in S3, so the disk set only shrinks —
+// the migration sweep (#550 Phase 2) drains it and the fallback then
+// never fires. Marker-free by construction.
+// - Staged writes SPOOL TO DISK (the same temp-in-blobs-dir shape as the
+// disk writer) and upload on commit. S3 has no rename, but a completed
+// PUT is atomic — an object is fetchable only in full — so "invisible
+// until commit" holds with no staging keys to garbage-collect. The spool
+// also gives the PUT a known ContentLength (R2 requires one).
+// - Reads are CHUNKED RANGED GETs behind an io.ReadSeekCloser:
+// http.ServeContent seeks (size probe, Range requests — video scrubs)
+// and then reads in 32KB slices; fetching per slice would be an S3
+// request per 32KB. The reader fetches aligned 512KB windows instead —
+// ~2 requests/MB served, and a scrub costs one window, not a full
+// download. (#836's lesson: request count is the S3 bill, watch it.)
+//
+// Config (instance env):
+//
+// LCHAT_BLOB_S3_BUCKET the switch — set = S3 store, unset = disk
+// LCHAT_BLOB_S3_PREFIX key prefix; default = the DB's basename (the
+// host), so a shared fleet bucket needs no
+// per-instance wiring
+// LCHAT_BLOB_S3_REGION optional ("auto" for R2)
+// LCHAT_BLOB_S3_ENDPOINT optional custom endpoint (R2's
+// https://<account>.r2.cloudflarestorage.com);
+// implies path-style addressing
+//
+// Credentials ride the standard AWS chain (AWS_ACCESS_KEY_ID /
+// AWS_SECRET_ACCESS_KEY in the instance env file — R2 issues S3-compatible
+// pairs).
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "io/fs"
+ "os"
+ "path"
+ "strings"
+ "time"
+
+ "github.com/aws/aws-sdk-go-v2/aws"
+ awsconfig "github.com/aws/aws-sdk-go-v2/config"
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+ "github.com/aws/smithy-go"
+)
+
+// s3ReadChunk is the ranged-GET window. 512KB ≈ 2 requests per MB served;
+// small enough that a video scrub's first paint isn't waiting on a huge
+// window, large enough that request count stays a rounding error on the bill.
+const s3ReadChunk = 512 << 10
+
+// s3BlobTimeout bounds each individual S3 call (not a whole serve — a member
+// streaming a long video holds the reader open far longer than any one GET).
+const s3BlobTimeout = 60 * time.Second
+
+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 blobStore
+ // spoolDir stages in-flight uploads — the disk store's dir, so the
+ // temp-name convention ("." never appears in a valid blob id) keeps
+ // staged bytes unfetchable exactly as before.
+ spoolDir string
+}
+
+// s3BlobConfig reads the env switch. Nil config = feature off (disk store).
+type s3BlobConfig struct {
+ Bucket string
+ Prefix string
+ Region string
+ Endpoint string
+}
+
+func s3BlobConfigFromEnv() *s3BlobConfig {
+ bucket := strings.TrimSpace(os.Getenv("LCHAT_BLOB_S3_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")),
+ }
+}
+
+// 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) {
+ var opts []func(*awsconfig.LoadOptions) error
+ if cfg.Region != "" {
+ opts = append(opts, awsconfig.WithRegion(cfg.Region))
+ }
+ awsCfg, err := awsconfig.LoadDefaultConfig(context.Background(), opts...)
+ if err != nil {
+ return nil, fmt.Errorf("blob s3: load aws config: %w", err)
+ }
+ client := awss3.NewFromConfig(awsCfg, func(o *awss3.Options) {
+ if cfg.Endpoint != "" {
+ o.BaseEndpoint = &cfg.Endpoint
+ // Custom endpoints (R2, test fakes) address by path, not vhost.
+ o.UsePathStyle = true
+ }
+ // The SDK's default-on flexible checksums wrap PutObject bodies in
+ // aws-chunked framing, which R2 rejects; when-required sends plain
+ // bodies (ContentLength is always known — the spool measured it).
+ o.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
+ o.ResponseChecksumValidation = aws.ResponseChecksumValidationWhenRequired
+ })
+ prefix := cfg.Prefix
+ if prefix == "" {
+ prefix = defaultPrefix
+ }
+ return &s3BlobStore{client: client, bucket: cfg.Bucket, prefix: prefix,
+ fallback: disk, spoolDir: disk.dir}, nil
+}
+
+func (s *s3BlobStore) key(id string) string { return path.Join(s.prefix, id) }
+
+// s3NotFound folds the three shapes a missing object answers with (NoSuchKey
+// on GET, NotFound on HEAD, and bare 404s from S3-compatibles) into one test.
+func s3NotFound(err error) bool {
+ var apiErr smithy.APIError
+ if errors.As(err, &apiErr) {
+ switch apiErr.ErrorCode() {
+ case "NoSuchKey", "NotFound", "404":
+ return true
+ }
+ }
+ return false
+}
+
+func (s *s3BlobStore) create() (blobWriter, error) {
+ f, err := os.CreateTemp(s.spoolDir, "up.*.part")
+ if err != nil {
+ return nil, err
+ }
+ return &s3BlobWriter{f: f, store: s}, nil
+}
+
+func (s *s3BlobStore) open(id string) (io.ReadSeekCloser, int64, time.Time, error) {
+ ctx, cancel := context.WithTimeout(context.Background(), s3BlobTimeout)
+ defer cancel()
+ key := s.key(id)
+ 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)
+ }
+ return nil, 0, time.Time{}, err
+ }
+ size := int64(0)
+ if head.ContentLength != nil {
+ size = *head.ContentLength
+ }
+ mod := time.Time{}
+ if head.LastModified != nil {
+ mod = *head.LastModified
+ }
+ return &s3Reader{store: s, key: key, size: size}, size, mod, nil
+}
+
+func (s *s3BlobStore) remove(id string) error {
+ ctx, cancel := context.WithTimeout(context.Background(), s3BlobTimeout)
+ defer cancel()
+ key := s.key(id)
+ if _, err := s.client.DeleteObject(ctx, &awss3.DeleteObjectInput{Bucket: &s.bucket, Key: &key}); err != nil && !s3NotFound(err) {
+ return err
+ }
+ // 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)
+}
+
+type s3BlobWriter struct {
+ f *os.File
+ store *s3BlobStore
+}
+
+func (w *s3BlobWriter) Write(p []byte) (int, error) { return w.f.Write(p) }
+
+func (w *s3BlobWriter) commit(id string) error {
+ defer os.Remove(w.f.Name())
+ size, err := w.f.Seek(0, io.SeekEnd)
+ if err != nil {
+ w.f.Close()
+ return err
+ }
+ if _, err := w.f.Seek(0, io.SeekStart); err != nil {
+ w.f.Close()
+ return err
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), s3BlobTimeout)
+ defer cancel()
+ key := w.store.key(id)
+ _, err = w.store.client.PutObject(ctx, &awss3.PutObjectInput{
+ Bucket: &w.store.bucket,
+ Key: &key,
+ Body: w.f,
+ ContentLength: &size,
+ })
+ w.f.Close()
+ return err
+}
+
+func (w *s3BlobWriter) abort() {
+ w.f.Close()
+ os.Remove(w.f.Name())
+}
+
+// s3Reader is a lazy io.ReadSeekCloser over one object: seeks are position
+// math (http.ServeContent's size probe costs nothing), reads fetch aligned
+// s3ReadChunk windows on demand.
+type s3Reader struct {
+ store *s3BlobStore
+ key string
+ size int64
+ pos int64
+ // The current window: buf holds object bytes [bufStart, bufStart+len(buf)).
+ buf []byte
+ bufStart int64
+}
+
+func (r *s3Reader) Read(p []byte) (int, error) {
+ if r.pos >= r.size {
+ return 0, io.EOF
+ }
+ // Miss (or first read): fetch the aligned window containing pos.
+ if r.buf == nil || r.pos < r.bufStart || r.pos >= r.bufStart+int64(len(r.buf)) {
+ start := (r.pos / s3ReadChunk) * s3ReadChunk
+ end := start + s3ReadChunk - 1 // Range is inclusive
+ if end >= r.size {
+ end = r.size - 1
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), s3BlobTimeout)
+ rng := fmt.Sprintf("bytes=%d-%d", start, end)
+ out, err := r.store.client.GetObject(ctx, &awss3.GetObjectInput{
+ Bucket: &r.store.bucket, Key: &r.key, Range: &rng,
+ })
+ if err != nil {
+ cancel()
+ if s3NotFound(err) {
+ // Deleted mid-serve (GC race) — surface as the fs error the
+ // handler already maps to 404.
+ return 0, fs.ErrNotExist
+ }
+ return 0, err
+ }
+ buf, err := io.ReadAll(out.Body)
+ out.Body.Close()
+ cancel()
+ if err != nil {
+ return 0, err
+ }
+ r.buf, r.bufStart = buf, start
+ }
+ n := copy(p, r.buf[r.pos-r.bufStart:])
+ r.pos += int64(n)
+ return n, nil
+}
+
+func (r *s3Reader) Seek(offset int64, whence int) (int64, error) {
+ var next int64
+ switch whence {
+ case io.SeekStart:
+ next = offset
+ case io.SeekCurrent:
+ next = r.pos + offset
+ case io.SeekEnd:
+ next = r.size + offset
+ default:
+ return 0, fmt.Errorf("blob s3: bad whence %d", whence)
+ }
+ if next < 0 {
+ return 0, fmt.Errorf("blob s3: negative seek")
+ }
+ r.pos = next
+ return next, nil
+}
+
+func (r *s3Reader) Close() error {
+ r.buf = nil
+ return nil
+}
diff --git a/s3blobstore_test.go b/s3blobstore_test.go
new file mode 100644
index 0000000..7cfcda0
--- /dev/null
+++ b/s3blobstore_test.go
@@ -0,0 +1,305 @@
+package main
+
+import (
+ "bytes"
+ "errors"
+ "fmt"
+ "io"
+ "io/fs"
+ "net/http"
+ "net/http/httptest"
+ "path/filepath"
+ "strconv"
+ "sync"
+ "testing"
+ "time"
+)
+
+// fakeS3 is just enough S3 for the store: path-style PUT/HEAD/GET(+Range)/
+// DELETE over an in-memory map, counting requests per verb so tests can pin
+// the request economy (#836's lesson: requests ARE the bill).
+type fakeS3 struct {
+ mu sync.Mutex
+ objects map[string][]byte // full path-style key: /bucket/prefix/id
+ gets int
+ puts int
+ heads int
+}
+
+func (f *fakeS3) handler(w http.ResponseWriter, r *http.Request) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ switch r.Method {
+ case http.MethodPut:
+ f.puts++
+ b, _ := io.ReadAll(r.Body)
+ f.objects[r.URL.Path] = b
+ w.WriteHeader(http.StatusOK)
+ case http.MethodHead:
+ f.heads++
+ b, ok := f.objects[r.URL.Path]
+ if !ok {
+ w.WriteHeader(http.StatusNotFound)
+ return
+ }
+ w.Header().Set("Content-Length", strconv.Itoa(len(b)))
+ w.Header().Set("Last-Modified", time.Now().UTC().Format(http.TimeFormat))
+ w.WriteHeader(http.StatusOK)
+ case http.MethodGet:
+ f.gets++
+ b, ok := f.objects[r.URL.Path]
+ if !ok {
+ w.Header().Set("Content-Type", "application/xml")
+ w.WriteHeader(http.StatusNotFound)
+ fmt.Fprint(w, `<Error><Code>NoSuchKey</Code></Error>`)
+ return
+ }
+ if rng := r.Header.Get("Range"); rng != "" {
+ var start, end int
+ if _, err := fmt.Sscanf(rng, "bytes=%d-%d", &start, &end); err == nil {
+ if end >= len(b) {
+ end = len(b) - 1
+ }
+ w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, end, len(b)))
+ w.WriteHeader(http.StatusPartialContent)
+ w.Write(b[start : end+1])
+ return
+ }
+ }
+ w.Write(b)
+ case http.MethodDelete:
+ delete(f.objects, r.URL.Path)
+ w.WriteHeader(http.StatusNoContent)
+ default:
+ w.WriteHeader(http.StatusNotImplemented)
+ }
+}
+
+// newTestS3Store wires an s3BlobStore at a fake endpoint over a real (temp)
+// disk store — the same construction openDB performs when the env is set.
+func newTestS3Store(t *testing.T) (*s3BlobStore, *fakeS3, *diskBlobStore) {
+ t.Helper()
+ fake := &fakeS3{objects: map[string][]byte{}}
+ srv := httptest.NewServer(http.HandlerFunc(fake.handler))
+ t.Cleanup(srv.Close)
+
+ // Static credentials + region so the SDK's signer is satisfied offline.
+ t.Setenv("AWS_ACCESS_KEY_ID", "test")
+ t.Setenv("AWS_SECRET_ACCESS_KEY", "test")
+ t.Setenv("AWS_REGION", "auto")
+
+ disk, err := newDiskBlobStore(filepath.Join(t.TempDir(), "blobs"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ store, err := newS3BlobStore(&s3BlobConfig{
+ Bucket: "test-bucket",
+ Endpoint: srv.URL,
+ }, disk, "space.example.com")
+ if err != nil {
+ t.Fatal(err)
+ }
+ return store, fake, disk
+}
+
+// The blobStore contract (TestDiskBlobStoreRoundTrip's sequence) against
+// implementation #2. Any drift between the two implementations here is a bug
+// in the new one — the contract came first.
+func TestS3BlobStoreRoundTrip(t *testing.T) {
+ store, fake, _ := newTestS3Store(t)
+ payload := []byte("pretend-this-is-encrypted-bytes\x00\x01\x02")
+
+ w, err := store.create()
+ if err != nil {
+ t.Fatalf("create: %v", err)
+ }
+ if _, err := w.Write(payload[:10]); err != nil {
+ t.Fatalf("write: %v", err)
+ }
+ // Staged bytes are invisible until commit — S3 empty, fallback empty.
+ if _, _, _, err := store.open("blobid_123-abc"); !errors.Is(err, fs.ErrNotExist) {
+ t.Fatalf("open before commit: want fs.ErrNotExist, got %v", err)
+ }
+ if _, err := w.Write(payload[10:]); err != nil {
+ t.Fatalf("write: %v", err)
+ }
+ if err := w.commit("blobid_123-abc"); err != nil {
+ t.Fatalf("commit: %v", err)
+ }
+ if fake.puts != 1 {
+ t.Fatalf("puts = %d, want exactly 1 (the commit)", fake.puts)
+ }
+ // The object landed under the derived per-instance prefix.
+ if _, ok := fake.objects["/test-bucket/space.example.com/blobid_123-abc"]; !ok {
+ t.Fatalf("object key wrong; have %v", keysOf(fake))
+ }
+
+ r, size, modTime, err := store.open("blobid_123-abc")
+ if err != nil {
+ t.Fatalf("open: %v", err)
+ }
+ got, err := io.ReadAll(r)
+ r.Close()
+ if err != nil || !bytes.Equal(got, payload) {
+ t.Fatalf("round-trip mismatch: err=%v got=%q", err, got)
+ }
+ if size != int64(len(payload)) {
+ t.Fatalf("size = %d, want %d", size, len(payload))
+ }
+ if modTime.IsZero() {
+ t.Fatal("zero modtime")
+ }
+
+ if err := store.remove("blobid_123-abc"); err != nil {
+ t.Fatalf("remove: %v", err)
+ }
+ if _, _, _, err := store.open("blobid_123-abc"); !errors.Is(err, fs.ErrNotExist) {
+ t.Fatalf("open after remove: want fs.ErrNotExist, got %v", err)
+ }
+ if err := store.remove("blobid_123-abc"); err != nil {
+ t.Fatalf("remove of missing id: %v", err)
+ }
+}
+
+func keysOf(f *fakeS3) []string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ out := make([]string, 0, len(f.objects))
+ for k := range f.objects {
+ out = append(out, k)
+ }
+ return out
+}
+
+// An aborted staged write publishes nothing and leaves no spool file behind.
+func TestS3BlobStoreAbortDiscards(t *testing.T) {
+ store, fake, disk := newTestS3Store(t)
+ w, err := store.create()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := w.Write([]byte("half an upload")); err != nil {
+ t.Fatal(err)
+ }
+ w.abort()
+ if len(keysOf(fake)) != 0 {
+ t.Fatalf("abort published: %v", keysOf(fake))
+ }
+ entries, err := filepath.Glob(filepath.Join(disk.dir, "*"))
+ if err != nil || len(entries) != 0 {
+ t.Fatalf("spool not cleaned: %v %v", entries, err)
+ }
+}
+
+// Pre-flip blobs live on disk; the s3 store serves and reaps them through its
+// fallback, so a flipped instance never strands its history.
+func TestS3BlobStoreFallbackServesDiskBlobs(t *testing.T) {
+ store, _, disk := newTestS3Store(t)
+ w, err := disk.create()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := w.Write([]byte("pre-flip bytes")); err != nil {
+ t.Fatal(err)
+ }
+ if err := w.commit("oldblob"); err != nil {
+ t.Fatal(err)
+ }
+
+ r, size, _, err := store.open("oldblob")
+ if err != nil {
+ t.Fatalf("open via fallback: %v", err)
+ }
+ got, _ := io.ReadAll(r)
+ r.Close()
+ if string(got) != "pre-flip bytes" || size != 14 {
+ t.Fatalf("fallback read = %q (%d)", got, size)
+ }
+
+ if err := store.remove("oldblob"); err != nil {
+ t.Fatalf("remove via fallback: %v", err)
+ }
+ if _, _, _, err := disk.open("oldblob"); !errors.Is(err, fs.ErrNotExist) {
+ t.Fatalf("disk bytes survived remove: %v", err)
+ }
+}
+
+// The reader's request economy: a size probe costs nothing, a scrub to the
+// middle of a video costs ONE window fetch, and a sequential read costs
+// ceil(size/window) — never a request per 32KB ServeContent slice.
+func TestS3ReaderRangedWindows(t *testing.T) {
+ store, fake, _ := newTestS3Store(t)
+ payload := bytes.Repeat([]byte("0123456789abcdef"), (s3ReadChunk*3)/16/2) // 1.5 windows
+ w, _ := store.create()
+ w.Write(payload)
+ if err := w.commit("video"); err != nil {
+ t.Fatal(err)
+ }
+
+ r, size, _, err := store.open("video")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer r.Close()
+ if size != int64(len(payload)) {
+ t.Fatalf("size = %d, want %d", size, len(payload))
+ }
+
+ // http.ServeContent's size probe: seeks only — zero GETs.
+ base := fake.gets
+ if _, err := r.Seek(0, io.SeekEnd); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.Seek(0, io.SeekStart); err != nil {
+ t.Fatal(err)
+ }
+ if fake.gets != base {
+ t.Fatalf("seeks cost %d GETs", fake.gets-base)
+ }
+
+ // A Range request landing mid-object (the video scrub): one window.
+ if _, err := r.Seek(int64(s3ReadChunk+100), io.SeekStart); err != nil {
+ t.Fatal(err)
+ }
+ slice := make([]byte, 64<<10)
+ if _, err := io.ReadFull(r, slice); err != nil {
+ t.Fatal(err)
+ }
+ if fake.gets-base != 1 {
+ t.Fatalf("mid-object 64KB read cost %d GETs, want 1", fake.gets-base)
+ }
+ if !bytes.Equal(slice, payload[s3ReadChunk+100:s3ReadChunk+100+64<<10]) {
+ t.Fatal("scrub read returned wrong bytes")
+ }
+
+ // Full sequential read from the top: exactly ceil(1.5 windows) = 2 GETs
+ // (the mid-object window above already cached one of them — reset first).
+ r2, _, _, err := store.open("video")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer r2.Close()
+ base = fake.gets
+ got, err := io.ReadAll(r2)
+ if err != nil || !bytes.Equal(got, payload) {
+ t.Fatalf("sequential read mismatch: %v", err)
+ }
+ if fake.gets-base != 2 {
+ t.Fatalf("sequential 1.5-window read cost %d GETs, want 2", fake.gets-base)
+ }
+}
+
+// The env switch: unset = disk store untouched; set = s3 store with the
+// prefix derived from the DB basename when none is named.
+func TestS3BlobConfigFromEnv(t *testing.T) {
+ t.Setenv("LCHAT_BLOB_S3_BUCKET", "")
+ if cfg := s3BlobConfigFromEnv(); cfg != nil {
+ t.Fatalf("empty bucket produced config %+v", cfg)
+ }
+ t.Setenv("LCHAT_BLOB_S3_BUCKET", "media")
+ t.Setenv("LCHAT_BLOB_S3_PREFIX", "/custom/")
+ cfg := s3BlobConfigFromEnv()
+ if cfg == nil || cfg.Bucket != "media" || cfg.Prefix != "custom" {
+ t.Fatalf("config = %+v", cfg)
+ }
+}