Writable VFS: FileSize race after sync flush causes "database disk image is malformed"

## Summary

After `syncToRemoteWithLock` uploads dirty pages and clears `f.dirty`, `FileSize()` can transiently return a size smaller than the actual database. SQLite compares `xFileSize` against the in-header page count (offset 28–31), sees a mismatch, and returns `SQLITE_CORRUPT` — surfacing as `"database disk image is malformed"` on the next read.

This reproduces on v0.5.11 with the writable VFS at roughly a 20% flake rate under a bulk-write + sync workload.

## Root cause

`FileSize()` computes the database size as `max(f.index ∪ f.pending ∪ f.dirty)`:

```go
// vfs.go — FileSize()
for pgno := range f.index { ... }
for pgno := range f.pending { ... }
for pgno := range f.dirty { ... }
```

`syncToRemoteWithLock()` uploads the dirty pages to S3, then clears `f.dirty`:

```go
// vfs.go — syncToRemoteWithLock()
f.dirty = make(map[uint32]int64)
```

It intentionally does **not** update `f.index`, relying on the poll goroutine to repopulate it on the next tick. Between the sync flush and the next poll, the synced pages are invisible to `FileSize()` — they're no longer in `f.dirty`, not yet in `f.index`, and never in `f.pending`. If those pages included the highest page numbers in the database, `FileSize()` returns a value smaller than `f.commit`, and SQLite reports corruption.

## Reproduction

The reproducer below uses MinIO at `localhost:9000`. Each of 30 trials creates a table, bulk-inserts 100 rows (growing to ~45 pages), deletes a few rows to exercise the freelist, waits for the sync ticker, then runs `SELECT count(*)`. On stock v0.5.11 roughly 6/30 trials fail.

Reproducer (cmd/repro/main.go)

```go
package main

import (
"context"
"database/sql"
"fmt"
"log/slog"
"os"
"strings"
"time"

"github.com/benbjohnson/litestream"
_ "github.com/benbjohnson/litestream/s3"
_ "github.com/mattn/go-sqlite3"
"github.com/psanford/sqlite3vfs"
)

const (
trials = 30
syncInterval = time.Second
replicaBucket = "s3://litestream-malformed-repro"
replicaQuery = "?endpoint=http://localhost:9000®ion=us-east-1&forcePathStyle=true"
)

func main() {
os.Setenv("AWS_ACCESS_KEY_ID", "minioadmin")
os.Setenv("AWS_SECRET_ACCESS_KEY", "minioadmin")

ctx := context.Background()
flakes := 0
for i := 1; i <= trials; i++ {
ok, err := runTrial(ctx, i)
if err != nil {
fmt.Printf("trial %02d: ERROR %v\n", i, err)
os.Exit(1)
}
if !ok {
flakes++
fmt.Printf("trial %02d: FLAKED — database disk image is malformed\n", i)
}
}
fmt.Printf("\n%d/%d trials flaked (%.1f%%)\n", flakes, trials, 100*float64(flakes)/float64(trials))
}

func runTrial(ctx context.Context, trial int) (bool, error) {
replicaURL := fmt.Sprintf("%s/trial-%02d%s", replicaBucket, trial, replicaQuery)
if c, err := litestream.NewReplicaClientFromURL(replicaURL); err == nil {
_ = c.Init(ctx)
_ = c.DeleteAll(ctx)
}
if err := phaseZero(ctx, trial, replicaURL); err != nil {
return false, fmt.Errorf("phase 0: %w", err)
}
return phaseOne(ctx, trial, replicaURL)
}

func phaseZero(ctx context.Context, trial int, replicaURL string) error {
db, err := openWritableVFS(ctx, fmt.Sprintf("seed-%02d", trial), replicaURL)
if err != nil {
return err
}
defer db.Close()
if _, err := db.ExecContext(ctx, `CREATE TABLE notes (id INTEGER PRIMARY KEY, body BLOB)`); err != nil {
return err
}
if err := bulkInsert(ctx, db, 1, 25); err != nil {
return err
}
time.Sleep(2 * syncInterval)
return nil
}

func phaseOne(ctx context.Context, trial int, replicaURL string) (bool, error) {
db, err := openWritableVFS(ctx, fmt.Sprintf("test-%02d", trial), replicaURL)
if err != nil {
return false, err
}
defer db.Close()
if err := bulkInsert(ctx, db, 26, 125); err != nil {
return false, err
}
if _, err := db.ExecContext(ctx, `DELETE FROM notes WHERE id <= 12`); err != nil {
return false, err
}
time.Sleep(2 * syncInterval)
var n int
err = db.QueryRowContext(ctx, `SELECT count(*) FROM notes`).Scan(&n)
if err == nil {
return true, nil
}
if strings.Contains(err.Error(), "database disk image is malformed") {
return false, nil
}
return false, err
}

func bulkInsert(ctx context.Context, db *sql.DB, from, to int) error {
pad := strings.Repeat("x", 1024)
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return err
}
stmt, err := tx.PrepareContext(ctx, `INSERT INTO notes (id, body) VALUES (?, ?)`)
if err != nil {
_ = tx.Rollback()
return err
}
for i := from; i <= to; i++ {
if _, err := stmt.ExecContext(ctx, i, pad); err != nil {
_ = stmt.Close()
_ = tx.Rollback()
return err
}
}
_ = stmt.Close()
return tx.Commit()
}

func openWritableVFS(ctx context.Context, vfsName, replicaURL string) (*sql.DB, error) {
client, err := litestream.NewReplicaClientFromURL(replicaURL)
if err != nil {
return nil, err
}
if err := client.Init(ctx); err != nil {
return nil, err
}
vfs := litestream.NewVFS(client, slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})))
vfs.WriteEnabled = true
vfs.WriteSyncInterval = syncInterval
if err := sqlite3vfs.RegisterVFS(vfsName, vfs); err != nil {
return nil, err
}
db, err := sql.Open("sqlite3", "file:db.sqlite?vfs="+vfsName)
if err != nil {
return nil, err
}
db.SetMaxOpenConns(1)
return db, nil
}
```

## Fix

Anchor `FileSize()` to `f.commit`, which tracks the highest page number written and is never cleared by sync. In `vfs.go`, `FileSize()`, after the existing loops:

```go
if v := int64(f.commit) * int64(pageSize); v > size {
size = v
}
```

This is safe because `f.commit` is updated on every `WriteAt` that extends the database and on `Truncate`, and it is protected by `f.mu` which `FileSize` already holds. With this change the reproducer passes 0/30 consistently.

## Environment

- Litestream v0.5.11 (`a590e05`)
- Go 1.25.1
- MinIO for S3 backend (but the bug is backend-agnostic — it's a VFS-layer race)

Start in vfs.go with FileSize(), then read syncToRemoteWithLock() and the handling of f.commit, f.index, f.pending, and f.dirty. Run cmd/repro/main.go against MinIO to reproduce the intermittent corruption, then verify that the writable VFS consistently reports a size at least as large as the committed database and the 30-trial reproducer completes without malformed-database errors.

Written by the indexing model from the issue text.