Keep v2 reads off whole-table scans found by a SQL plan audit

cmd/read.go

@@ -2,6 +2,7 @@ package cmd
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -45,9 +46,11 @@ func openCommandReadSource(
 	}
 	if mode.Primary {
 		storePath := filepath.Join(dataDir, "v2", "store.sqlite3")
-		store, err := sqlite.Open(storePath)
+		// A reader never migrates the store the app owns: a migration holds
+		// the write lock for its whole run and would stall the daemon's writes.
+		store, err := sqlite.OpenWithoutMigrating(storePath)
 		if err != nil {
-			return nil, fmt.Errorf("open v2 read store %q: %w", storePath, err)
+			return nil, fmt.Errorf("open v2 read store %q: %w", storePath, clientStoreOpenError(err))
 		}
 		if banner != nil {
 			fmt.Fprintln(banner, "reading v2 store")
@@ -173,3 +176,12 @@ func firstNonEmpty(vals ...string) string {
 func parseDayBound(s string, endOfDay bool) (int64, error) {
 	return db.ParseDayBound(s, endOfDay)
 }
+
+// clientStoreOpenError explains the refusal a read client gets when this
+// binary is newer than the store the app last upgraded.
+func clientStoreOpenError(err error) error {
+	if errors.Is(err, sqlite.ErrMigrationPending) {
+		return fmt.Errorf("%w (quit and reopen the OpenMessage app so it upgrades the store, then retry)", err)
+	}
+	return err
+}

cmd/read_test.go

@@ -2,8 +2,13 @@ package cmd
 
 import (
 	"bytes"
+	"database/sql"
+	"errors"
+	"fmt"
+	"io"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -210,3 +215,78 @@ func TestOpenCommandReadSourceLegacyDoesNotRepairStore(t *testing.T) {
 		t.Fatal("openCommandReadSource ran the startup repair sweeps: the legacy reaction placeholder was repaired away")
 	}
 }
+
+// TestOpenCommandReadSourceNeverMigratesTheV2Store pins the other half of the
+// read-only contract: `read` and `status` from a binary newer than the app's
+// store refuse it instead of migrating it. A migration holds SQLite's write
+// lock for its whole run, so applying one under the running app would stall
+// (and, past busy_timeout, fail) the daemon's inbox appends.
+func TestOpenCommandReadSourceNeverMigratesTheV2Store(t *testing.T) {
+	dataDir := t.TempDir()
+	t.Setenv("OPENMESSAGES_DATA_DIR", dataDir)
+	t.Setenv("OPENMESSAGES_DEMO", "0")
+	t.Setenv("OPENMESSAGES_APP_SANDBOX", "1")
+	t.Setenv("OPENMESSAGES_V2_PRIMARY", "1")
+	t.Setenv("OPENMESSAGES_V2_SEND", "")
+	t.Setenv("OPENMESSAGES_V2_INGEST", "")
+
+	v2Dir := filepath.Join(dataDir, "v2")
+	if err := os.MkdirAll(v2Dir, 0o700); err != nil {
+		t.Fatalf("create v2 dir: %v", err)
+	}
+	storePath := filepath.Join(v2Dir, "store.sqlite3")
+	store, err := sqlite.Open(storePath)
+	if err != nil {
+		t.Fatalf("sqlite.Open(): %v", err)
+	}
+	if err := store.Close(); err != nil {
+		t.Fatal(err)
+	}
+	latest := rollBackLedgerOneVersion(t, storePath)
+
+	session, err := openCommandReadSource(zerolog.Nop(), io.Discard)
+	if err == nil {
+		session.Close()
+		t.Fatal("openCommandReadSource() opened a store with a pending migration; want a refusal")
+	}
+	if !errors.Is(err, sqlite.ErrMigrationPending) || !strings.Contains(err.Error(), "reopen the OpenMessage app") {
+		t.Fatalf("openCommandReadSource() error = %v, want ErrMigrationPending telling the user to reopen the app", err)
+	}
+	if got := ledgerVersionCount(t, storePath); got != latest-1 {
+		t.Fatalf("ledger rows = %d after the refused read, want %d: the reader applied a migration", got, latest-1)
+	}
+}
+
+// rollBackLedgerOneVersion removes the newest ledger row and lowers
+// user_version to match: the shape of a store the running app has not
+// upgraded to this build's schema yet. It returns the version it removed.
+func rollBackLedgerOneVersion(t *testing.T, storePath string) int {
+	t.Helper()
+	latest := ledgerVersionCount(t, storePath)
+	database, err := sql.Open("sqlite", storePath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer database.Close()
+	if _, err := database.Exec(`DELETE FROM schema_migrations WHERE version = ?`, latest); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := database.Exec(fmt.Sprintf(`PRAGMA user_version = %d`, latest-1)); err != nil {
+		t.Fatal(err)
+	}
+	return latest
+}
+
+func ledgerVersionCount(t *testing.T, storePath string) int {
+	t.Helper()
+	database, err := sql.Open("sqlite", storePath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer database.Close()
+	var rows int
+	if err := database.QueryRow(`SELECT COUNT(*) FROM schema_migrations`).Scan(&rows); err != nil {
+		t.Fatal(err)
+	}
+	return rows
+}

cmd/repair.go

@@ -100,9 +100,17 @@ func runRepairGoogleIDSpace(ctx context.Context, logger zerolog.Logger, args []s
 		defer lock.Close()
 	}
 
-	store, err := sqlite.Open(storePath)
+	// Only an exclusive owner migrates the store: --apply holds the instance
+	// lock and has checked the daemon is down. A dry run may run beside the
+	// app, and a migration's write lock would stall the daemon's writes.
+	var store *sqlite.Store
+	if apply {
+		store, err = sqlite.Open(storePath)
+	} else {
+		store, err = sqlite.OpenWithoutMigrating(storePath)
+	}
 	if err != nil {
-		return fmt.Errorf("open v2 store: %w", err)
+		return fmt.Errorf("open v2 store: %w", clientStoreOpenError(err))
 	}
 	defer store.Close()
 	messages, err := sqlite.NewMessageRepository(store, time.Now)
@@ -115,11 +123,12 @@ func runRepairGoogleIDSpace(ctx context.Context, logger zerolog.Logger, args []s
 		if _, err := os.Stat(reference); err != nil {
 			return fmt.Errorf("reference store %q: %w", reference, err)
 		}
-		referenceStore, err = sqlite.Open(reference)
+		var closeReference func()
+		referenceStore, closeReference, err = openReferenceCopy(reference)
 		if err != nil {
 			return fmt.Errorf("open reference store: %w", err)
 		}
-		defer referenceStore.Close()
+		defer closeReference()
 	}
 
 	report, err := ingest.PlanGoogleIDSpaceRepair(ctx, store, messages, ingest.IDSpaceRepairOptions{
@@ -228,6 +237,30 @@ func daemonAnswers(probeURL string) bool {
 	return response.StatusCode == http.StatusOK
 }
 
+// openReferenceCopy opens a copy of the reference snapshot, migrated to this
+// build's schema, and returns a func that closes it and removes the copy. The
+// snapshot is often older than the build, and migrating it in place would
+// rewrite the user's backup.
+func openReferenceCopy(reference string) (*sqlite.Store, func(), error) {
+	dir, err := os.MkdirTemp("", "openmessage-repair-reference-")
+	if err != nil {
+		return nil, nil, err
+	}
+	if err := copyStoreFiles(reference, dir); err != nil {
+		_ = os.RemoveAll(dir)
+		return nil, nil, fmt.Errorf("copy reference store: %w", err)
+	}
+	store, err := sqlite.Open(filepath.Join(dir, filepath.Base(reference)))
+	if err != nil {
+		_ = os.RemoveAll(dir)
+		return nil, nil, err
+	}
+	return store, func() {
+		_ = store.Close()
+		_ = os.RemoveAll(dir)
+	}, nil
+}
+
 func copyStoreFiles(storePath, backupDir string) error {
 	if err := os.MkdirAll(backupDir, 0o700); err != nil {
 		return err

cmd/repair_test.go

@@ -0,0 +1,104 @@
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/rs/zerolog"
+
+	"github.com/maxghenis/openmessage/internal/storage/sqlite"
+)
+
+// seedRepairStore provisions a current v2 store with the Google account the
+// repair plans for, and returns its path.
+func seedRepairStore(t *testing.T) string {
+	t.Helper()
+	dataDir := t.TempDir()
+	t.Setenv("OPENMESSAGES_DATA_DIR", dataDir)
+	t.Setenv("OPENMESSAGES_DEMO", "0")
+	t.Setenv("OPENMESSAGES_APP_SANDBOX", "1")
+	v2Dir := filepath.Join(dataDir, "v2")
+	if err := os.MkdirAll(v2Dir, 0o700); err != nil {
+		t.Fatalf("create v2 dir: %v", err)
+	}
+	storePath := filepath.Join(v2Dir, "store.sqlite3")
+	store, err := sqlite.Open(storePath)
+	if err != nil {
+		t.Fatalf("sqlite.Open(): %v", err)
+	}
+	nowMS := time.Now().UnixMilli()
+	if err := store.UpsertAccount(sqlite.Account{
+		AccountID:   "google-primary",
+		BridgeKey:   "google_messages",
+		Mode:        sqlite.AccountModeLive,
+		Enabled:     true,
+		ConfigJSON:  "{}",
+		CreatedAtMS: nowMS,
+		UpdatedAtMS: nowMS,
+	}); err != nil {
+		store.Close()
+		t.Fatalf("UpsertAccount(): %v", err)
+	}
+	if err := store.Close(); err != nil {
+		t.Fatal(err)
+	}
+	return storePath
+}
+
+// A dry run needs no exclusive ownership (no daemon probe, no instance lock),
+// so it must never migrate: run beside the app, a migration's write lock
+// would stall the daemon's writes.
+func TestRepairDryRunNeverMigratesTheStore(t *testing.T) {
+	storePath := seedRepairStore(t)
+	latest := rollBackLedgerOneVersion(t, storePath)
+
+	var stdout, stderr bytes.Buffer
+	err := runRepairGoogleIDSpace(context.Background(), zerolog.Nop(), []string{"--since", "1"}, &stdout, &stderr)
+	if !errors.Is(err, sqlite.ErrMigrationPending) || !strings.Contains(err.Error(), "reopen the OpenMessage app") {
+		t.Fatalf("dry run error = %v, want ErrMigrationPending telling the user to reopen the app", err)
+	}
+	if got := ledgerVersionCount(t, storePath); got != latest-1 {
+		t.Fatalf("ledger rows = %d after the dry run, want %d: the dry run applied a migration", got, latest-1)
+	}
+}
+
+// The reference snapshot is the user's own backup. The repair reads a migrated
+// copy of it and never writes the file itself, so a read-only snapshot works
+// and stays byte-identical.
+func TestRepairDryRunReadsACopyOfTheReferenceSnapshot(t *testing.T) {
+	storePath := seedRepairStore(t)
+	reference := filepath.Join(t.TempDir(), "reference.sqlite3")
+	payload, err := os.ReadFile(storePath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(reference, payload, 0o400); err != nil {
+		t.Fatal(err)
+	}
+
+	var stdout, stderr bytes.Buffer
+	if err := runRepairGoogleIDSpace(
+		context.Background(), zerolog.Nop(),
+		[]string{"--since", "1", "--reference", reference, "--json"}, &stdout, &stderr,
+	); err != nil {
+		t.Fatalf("dry run with a read-only reference: %v\n%s", err, stderr.String())
+	}
+	after, err := os.ReadFile(reference)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Equal(payload, after) {
+		t.Fatal("the repair wrote the reference snapshot")
+	}
+	for _, suffix := range []string{"-wal", "-shm"} {
+		if _, err := os.Stat(reference + suffix); !errors.Is(err, os.ErrNotExist) {
+			t.Fatalf("the repair created %s next to the reference snapshot: %v", reference+suffix, err)
+		}
+	}
+}

cmd/serve.go

@@ -881,9 +881,11 @@ func runServeMCPClient(logger zerolog.Logger, opts serveOptions) error {
 		if _, err := os.Stat(v2StorePath); err != nil {
 			return fmt.Errorf("v2-primary reads selected but no migrated store at %s: %w", v2StorePath, err)
 		}
-		opened, err := sqlite.Open(v2StorePath)
+		// Never migrate here: the daemon owns the store, and a migration's
+		// write lock would stall its writes for the whole index build.
+		opened, err := sqlite.OpenWithoutMigrating(v2StorePath)
 		if err != nil {
-			return fmt.Errorf("open v2 read store %q: %w", v2StorePath, err)
+			return fmt.Errorf("open v2 read store %q: %w", v2StorePath, clientStoreOpenError(err))
 		}
 		v2Store = opened
 		defer func() {

cmd/serve_client_test.go

@@ -3,6 +3,7 @@ package cmd
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
@@ -204,6 +205,55 @@ func TestRunServeMCPClientAdoptsDaemonTruth(t *testing.T) {
 	}
 }
 
+// TestRunServeMCPClientNeverMigratesTheV2Store pins that the per-session MCP
+// client refuses a store the running app has not upgraded instead of
+// migrating it: the migration's write lock would stall the daemon's writes.
+func TestRunServeMCPClientNeverMigratesTheV2Store(t *testing.T) {
+	dataDir := t.TempDir()
+	v2Dir := filepath.Join(dataDir, "v2")
+	if err := os.MkdirAll(v2Dir, 0o700); err != nil {
+		t.Fatalf("create v2 dir: %v", err)
+	}
+	storePath := filepath.Join(v2Dir, "store.sqlite3")
+	store, err := sqlite.Open(storePath)
+	if err != nil {
+		t.Fatalf("provision v2 store: %v", err)
+	}
+	if err := store.Close(); err != nil {
+		t.Fatalf("close provisioned v2 store: %v", err)
+	}
+	latest := rollBackLedgerOneVersion(t, storePath)
+
+	daemon := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/api/status" {
+			http.NotFound(w, r)
+			return
+		}
+		json.NewEncoder(w).Encode(map[string]any{
+			"connected":  true,
+			"v2_primary": true,
+			"v2_send":    true,
+			"auth":       map[string]any{"data_dir": dataDir},
+		})
+	}))
+	defer daemon.Close()
+	daemonURL, err := url.Parse(daemon.URL)
+	if err != nil {
+		t.Fatalf("parse daemon URL: %v", err)
+	}
+	setClientModeTestEnv(t, dataDir)
+	t.Setenv("OPENMESSAGES_PORT", daemonURL.Port())
+
+	var logs syncBuffer
+	err = RunServe(zerolog.New(&logs), "--mcp-stdio")
+	if !errors.Is(err, sqlite.ErrMigrationPending) || !strings.Contains(err.Error(), "reopen the OpenMessage app") {
+		t.Fatalf("RunServe(--mcp-stdio) error = %v, want ErrMigrationPending telling the user to reopen the app\n%s", err, logs.String())
+	}
+	if got := ledgerVersionCount(t, storePath); got != latest-1 {
+		t.Fatalf("ledger rows = %d after the refused client start, want %d: the client applied a migration", got, latest-1)
+	}
+}
+
 // TestRunServeMCPClientAdoptsDaemonDataDir verifies that with no
 // OPENMESSAGES_DATA_DIR pinned, the client adopts the data directory the
 // running daemon reports, so an env-less MCP spawn serves the app's store