diff --git a/docs/PERSISTENCE.md b/docs/PERSISTENCE.md index 84b780e..fddd654 100644 --- a/docs/PERSISTENCE.md +++ b/docs/PERSISTENCE.md @@ -194,6 +194,21 @@ loop closes the current rows; cancellation and query/iteration failures are returned through the iterator. Ordinary `Read(ctx, from, 0)` still reads without a result limit. +Use `sqlite.OpenReadOnly(path, options...)` to read an existing store without +creating a database, migrating its schema, or changing its journal mode. The +stored schema version must be supported; an uninitialized or incompatible +database is rejected. `WithAutoMigrate(true)` is an error with this constructor. +`Append`, `SaveOffset`, `SaveSnapshot`, and `TruncateBefore` return +`sqlite.ErrReadOnly`. Ordinary `New` behavior is unchanged. + +Read-only stores use SQLite `mode=ro`, so later commits from a concurrent WAL +writer remain visible. They do not use `immutable=1` and do not provide a fixed +snapshot. SQLite may create or update WAL shared-memory sidecars when directory +permissions allow it. Schema validation happens at open; corruption encountered +in event pages is reported by reads, without an automatic integrity scan or +repair. Applications requiring an entirely immutable filesystem copy must +arrange that separately. + ### Durable-Streams Store Remote storage using the [Durable Streams](https://electric-sql.com/blog/2025/12/09/announcing-durable-streams) protocol - an HTTP-based persistent stream primitive for reliable, resumable, real-time data streaming developed by [Electric](https://electric-sql.com). diff --git a/stores/sqlite/options.go b/stores/sqlite/options.go index 1a540b3..08d6116 100644 --- a/stores/sqlite/options.go +++ b/stores/sqlite/options.go @@ -31,6 +31,7 @@ type config struct { path string busyTimeout time.Duration autoMigrate bool + readOnly bool logger Logger metricsHook MetricsHook streamBatchSize int @@ -54,7 +55,7 @@ func WithBusyTimeout(timeout time.Duration) Option { } // WithAutoMigrate enables or disables automatic schema migration -// Default is true +// Default is true for New; OpenReadOnly defaults to false and rejects true. func WithAutoMigrate(enabled bool) Option { return func(c *config) { c.autoMigrate = enabled diff --git a/stores/sqlite/readonly.go b/stores/sqlite/readonly.go new file mode 100644 index 0000000..65ef361 --- /dev/null +++ b/stores/sqlite/readonly.go @@ -0,0 +1,79 @@ +package sqlite + +import ( + "context" + "errors" + "fmt" + "net/url" + "path/filepath" +) + +// ErrReadOnly is returned by mutations on a store opened with OpenReadOnly. +var ErrReadOnly = errors.New("sqlite: store is read-only") + +// OpenReadOnly opens an existing file-backed store without changing its data, +// schema or journal mode. It never creates or migrates a database. The stored +// schema must match the version this package reads; WithAutoMigrate(true) is +// rejected. Other options have the same meaning as for New. +// +// SQLite mode=ro is used, not immutable=1: reads can observe later commits by +// a concurrent writer. SQLite may create or update WAL shared-memory sidecars +// when the directory permits it; this is not an immutable filesystem snapshot. +// Open validates the schema, while corruption encountered reading event pages +// is reported by the read operation. No integrity scan or repair is performed. +func OpenReadOnly(path string, opts ...Option) (*SQLiteStore, error) { + if path == "" || path == ":memory:" { + return nil, errors.New("sqlite: read-only opening requires an existing file path") + } + cfg := defaultConfig() + cfg.autoMigrate = false + cfg.readOnly = true + for _, opt := range opts { + opt(cfg) + } + if cfg.autoMigrate { + return nil, errors.New("sqlite: read-only opening cannot enable automatic migration") + } + abs, err := filepath.Abs(path) + if err != nil { + return nil, fmt.Errorf("sqlite: resolve read-only path: %w", err) + } + cfg.path = abs + uri := url.URL{Scheme: "file", Path: abs} + query := url.Values{"mode": {"ro"}, "_pragma": { + fmt.Sprintf("busy_timeout(%d)", cfg.busyTimeout.Milliseconds()), + "query_only(1)", + }} + uri.RawQuery = query.Encode() + db, err := dbOpener("sqlite", uri.String()) + if err != nil { + return nil, fmt.Errorf("sqlite: open read-only database: %w", err) + } + var version int + if err := db.QueryRowContext(context.Background(), "SELECT MAX(version) FROM schema_version").Scan(&version); err != nil { + _ = db.Close() + return nil, fmt.Errorf("sqlite: inspect read-only schema: %w", err) + } + if version != currentSchemaVersion { + _ = db.Close() + return nil, fmt.Errorf("sqlite: unsupported read-only schema version %d (expected %d)", version, currentSchemaVersion) + } + // database/sql Prepare may defer SQL compilation until first use. Execute + // bounded read queries so a missing table or column fails opening rather + // than appearing to be a valid initialized store. No event pages are scanned. + for _, query := range []string{ + "SELECT position, type, data, timestamp, event_id, origin, metadata FROM events LIMIT 0", + "SELECT subscription_id, position, updated_at FROM subscription_positions LIMIT 0", + "SELECT snapshot_id, position, data, updated_at FROM snapshots LIMIT 0", + } { + rows, err := db.QueryContext(context.Background(), query) + if err == nil { + err = rows.Close() + } + if err != nil { + _ = db.Close() + return nil, fmt.Errorf("sqlite: inspect read-only schema: %w", err) + } + } + return newFromDB(db, cfg, nil) +} diff --git a/stores/sqlite/readonly_test.go b/stores/sqlite/readonly_test.go new file mode 100644 index 0000000..02d0352 --- /dev/null +++ b/stores/sqlite/readonly_test.go @@ -0,0 +1,356 @@ +package sqlite + +import ( + "bytes" + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "reflect" + "testing" + "time" + + eventbus "github.com/jilio/ebu" +) + +func readOnlyFixture(t *testing.T) (string, []eventbus.Offset) { + t.Helper() + path := filepath.Join(t.TempDir(), "events.db") + writer, err := New(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = writer.Close() }) + var offsets []eventbus.Offset + for i := range 5 { + offset, err := writer.Append(t.Context(), &eventbus.Event{ + ID: fmt.Sprintf("event-%d", i), Type: "example", Data: []byte(fmt.Sprint(i)), + Timestamp: time.Unix(int64(i), 0).UTC(), Origin: "writer", Metadata: map[string]string{"example": "metadata"}, + }) + if err != nil { + t.Fatal(err) + } + offsets = append(offsets, offset) + } + if err := writer.SaveOffset(t.Context(), "reader", offsets[1]); err != nil { + t.Fatal(err) + } + if err := writer.SaveSnapshot(t.Context(), "projection", offsets[1], json.RawMessage(`{"count":2}`)); err != nil { + t.Fatal(err) + } + if err := writer.Close(); err != nil { + t.Fatal(err) + } + return path, offsets +} + +func rawReadOnlyFixtureSQL(t *testing.T, path, statement string) { + t.Helper() + db, err := sql.Open("sqlite", path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + if _, err := db.Exec(statement); err != nil { + t.Fatal(err) + } +} + +func readOnlyFileBytes(t *testing.T, path string) []byte { + t.Helper() + raw, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + return raw +} + +func TestOpenReadOnlyReadsWithoutDataSchemaOrJournalModeChanges(t *testing.T) { + for _, journalMode := range []string{"WAL", "DELETE"} { + t.Run(journalMode, func(t *testing.T) { + path, offsets := readOnlyFixture(t) + rawReadOnlyFixtureSQL(t, path, "PRAGMA journal_mode="+journalMode) + before := readOnlyFileBytes(t, path) + hook := &testMetricsHook{} + reader, err := OpenReadOnly(path, WithAutoMigrate(false), WithBusyTimeout(time.Second), WithStreamBatchSize(2), WithMetricsHook(hook)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = reader.Close() }) + events, next, err := reader.Read(t.Context(), eventbus.OffsetOldest, 2) + if err != nil || len(events) != 2 || next != offsets[1] || events[0].ID != "event-0" || events[0].Metadata["example"] != "metadata" { + t.Fatalf("first read: events=%v next=%q err=%v", events, next, err) + } + var streamed []string + for event, err := range reader.ReadStream(t.Context(), next) { + if err != nil { + t.Fatal(err) + } + streamed = append(streamed, event.ID) + } + if !reflect.DeepEqual(streamed, []string{"event-2", "event-3", "event-4"}) { + t.Fatalf("stream=%v", streamed) + } + if events, tail, err := reader.Read(t.Context(), eventbus.OffsetNewest, 0); err != nil || len(events) != 0 || tail != offsets[4] { + t.Fatalf("opaque tail=%q events=%v err=%v", tail, events, err) + } + if offset, err := reader.LoadOffset(t.Context(), "reader"); err != nil || offset != offsets[1] { + t.Fatalf("saved offset=%q %v", offset, err) + } + if _, found, err := reader.LookupOffset(t.Context(), "missing"); err != nil || found { + t.Fatalf("missing offset found=%v err=%v", found, err) + } + if offset, blob, err := reader.LoadSnapshot(t.Context(), "projection"); err != nil || offset != offsets[1] || string(blob) != `{"count":2}` { + t.Fatalf("snapshot=%q at=%q err=%v", blob, offset, err) + } + for _, mutate := range []func() error{ + func() error { _, err := reader.Append(t.Context(), nil); return err }, + func() error { return reader.SaveOffset(t.Context(), "reader", eventbus.OffsetNewest) }, + func() error { + return reader.SaveSnapshot(t.Context(), "projection", offsets[4], json.RawMessage(`{"changed":true}`)) + }, + func() error { _, err := reader.TruncateBefore(t.Context(), offsets[4]); return err }, + func() error { _, err := reader.TruncateBefore(t.Context(), eventbus.OffsetOldest); return err }, + } { + if err := mutate(); !errors.Is(err, ErrReadOnly) { + t.Fatalf("mutation returned %v", err) + } + } + if hook.appendCount != 1 || !errors.Is(hook.lastAppendErr, ErrReadOnly) || hook.saveOffsetCount != 1 || !errors.Is(hook.lastSaveOffErr, ErrReadOnly) { + t.Fatal("mutation rejection was not reported to metrics hook") + } + // The underlying SQLite connection must enforce mode=ro even if a + // future internal call accidentally bypasses the public method guard. + reader.db.SetMaxOpenConns(1) + if _, err := reader.db.Exec("PRAGMA query_only=OFF"); err != nil { + t.Fatal(err) + } + for _, statement := range []string{"CREATE TABLE unexpected (value TEXT)", "DELETE FROM events", "UPDATE schema_version SET version=99"} { + if _, err := reader.db.Exec(statement); err == nil { + t.Fatalf("read-only SQLite accepted %s", statement) + } + } + if err := reader.Close(); err != nil { + t.Fatal(err) + } + if after := readOnlyFileBytes(t, path); !bytes.Equal(before, after) { + t.Fatal("read-only open/read/rejected mutation/close changed database bytes") + } + if journalMode == "DELETE" { + if _, err := os.Stat(path + "-wal"); !os.IsNotExist(err) { + t.Fatal("read-only opening created WAL for a rollback-journal database") + } + } + }) + } +} + +func TestOpenReadOnlyRejectsMissingUninitializedCorruptAndUnsupportedWithoutRepair(t *testing.T) { + for _, kind := range []string{"missing", "empty", "garbage", "truncated", "unrelated", "missing-schema-version", "empty-schema-version", "older-schema", "future-schema", "missing-events", "missing-offsets", "missing-snapshots", "missing-event-column"} { + t.Run(kind, func(t *testing.T) { + path := filepath.Join(t.TempDir(), "events.db") + switch kind { + case "missing": + case "empty", "garbage": + raw := []byte(nil) + if kind == "garbage" { + raw = []byte("not a SQLite database") + } + if err := os.WriteFile(path, raw, 0o600); err != nil { + t.Fatal(err) + } + case "unrelated": + rawReadOnlyFixtureSQL(t, path, "CREATE TABLE unrelated (value TEXT)") + default: + path, _ = readOnlyFixture(t) + switch kind { + case "truncated": + if err := os.Truncate(path, 64); err != nil { + t.Fatal(err) + } + case "missing-schema-version": + rawReadOnlyFixtureSQL(t, path, "DROP TABLE schema_version") + case "empty-schema-version": + rawReadOnlyFixtureSQL(t, path, "DELETE FROM schema_version") + case "older-schema": + rawReadOnlyFixtureSQL(t, path, "DELETE FROM schema_version WHERE version=4") + case "future-schema": + rawReadOnlyFixtureSQL(t, path, "INSERT INTO schema_version(version) VALUES(5)") + case "missing-events": + rawReadOnlyFixtureSQL(t, path, "DROP TABLE events") + case "missing-offsets": + rawReadOnlyFixtureSQL(t, path, "DROP TABLE subscription_positions") + case "missing-snapshots": + rawReadOnlyFixtureSQL(t, path, "DROP TABLE snapshots") + case "missing-event-column": + rawReadOnlyFixtureSQL(t, path, "ALTER TABLE events DROP COLUMN metadata") + } + } + var before []byte + if kind != "missing" { + before = readOnlyFileBytes(t, path) + } + reader, err := OpenReadOnly(path) + if err == nil || reader != nil { + if reader != nil { + _ = reader.Close() + } + t.Fatal("invalid database opened read-only successfully") + } + if kind == "missing" { + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Fatal("read-only open created missing database") + } + } else if after := readOnlyFileBytes(t, path); !bytes.Equal(before, after) { + t.Fatal("failed read-only opening repaired or mutated database") + } + }) + } +} + +func TestOpenReadOnlyRejectsWriteOptionsAndEscapesLiteralFilenames(t *testing.T) { + for _, path := range []string{"", ":memory:"} { + if reader, err := OpenReadOnly(path); err == nil || reader != nil { + t.Fatalf("read-only path %q accepted", path) + } + } + missing := filepath.Join(t.TempDir(), "must-not-create.db") + if reader, err := OpenReadOnly(missing, WithAutoMigrate(true)); err == nil || reader != nil { + t.Fatal("write-enabling option accepted") + } + if _, err := os.Stat(missing); !os.IsNotExist(err) { + t.Fatal("conflicting options created a database") + } + path, _ := readOnlyFixture(t) + literal := filepath.Join(filepath.Dir(path), "literal%3fmode=rw?mode=rwc#fragment.db") + if err := os.Rename(path, literal); err != nil { + t.Fatal(err) + } + reader, err := OpenReadOnly(literal) + if err != nil { + t.Fatal(err) + } + defer reader.Close() + if events, _, err := reader.Read(t.Context(), eventbus.OffsetOldest, 0); err != nil || len(events) != 5 { + t.Fatalf("escaped path read: %d %v", len(events), err) + } + if _, err := reader.Append(t.Context(), nil); !errors.Is(err, ErrReadOnly) { + t.Fatalf("append with default options: %v", err) + } + if err := reader.SaveOffset(t.Context(), "reader", eventbus.OffsetOldest); !errors.Is(err, ErrReadOnly) { + t.Fatalf("save offset with default options: %v", err) + } +} + +func TestOpenReadOnlyReportsPathAndDriverErrors(t *testing.T) { + t.Run("removed-working-directory", func(t *testing.T) { + directory := t.TempDir() + t.Chdir(directory) + if err := os.Remove(directory); err != nil { + t.Fatal(err) + } + if reader, err := OpenReadOnly("relative.db"); err == nil || reader != nil { + t.Fatal("unresolvable relative path was accepted") + } + }) + t.Run("driver-open", func(t *testing.T) { + previous := dbOpener + t.Cleanup(func() { dbOpener = previous }) + want := errors.New("driver open failed") + dbOpener = func(string, string) (*sql.DB, error) { return nil, want } + if reader, err := OpenReadOnly(filepath.Join(t.TempDir(), "events.db")); !errors.Is(err, want) || reader != nil { + t.Fatalf("driver failure: reader=%v error=%v", reader, err) + } + }) +} + +func TestOpenReadOnlyObservesConcurrentWALCommitsAndPooledReadOnlyConnections(t *testing.T) { + path := filepath.Join(t.TempDir(), "events.db") + writer, err := New(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = writer.Close() }) + writer.db.SetMaxOpenConns(1) + if _, err := writer.db.Exec("PRAGMA wal_autocheckpoint=0"); err != nil { + t.Fatal(err) + } + reader, err := OpenReadOnly(path, WithStreamBatchSize(2)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = reader.Close() }) + if events, _, err := reader.Read(t.Context(), eventbus.OffsetOldest, 0); err != nil || len(events) != 0 { + t.Fatalf("initialized empty store: events=%v err=%v", events, err) + } + reader.db.SetMaxOpenConns(3) + var connections []*sql.Conn + for range 3 { + connection, err := reader.db.Conn(t.Context()) + if err != nil { + t.Fatal(err) + } + connections = append(connections, connection) + } + for _, connection := range connections { + var queryOnly int + if err := connection.QueryRowContext(t.Context(), "PRAGMA query_only").Scan(&queryOnly); err != nil || queryOnly != 1 { + t.Fatalf("pooled query_only=%d err=%v", queryOnly, err) + } + if _, err := connection.ExecContext(t.Context(), "DELETE FROM events"); err == nil { + t.Fatal("pooled connection was writable") + } + if err := connection.Close(); err != nil { + t.Fatal(err) + } + } + ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second) + committed := make(chan error) + done := make(chan struct{}) + defer func() { cancel(); <-done }() + go func() { + defer close(done) + for i := range 30 { + _, err := writer.Append(ctx, &eventbus.Event{ID: fmt.Sprintf("concurrent-%d", i), Type: "event", Data: []byte("payload"), Timestamp: time.Now()}) + select { + case committed <- err: + case <-ctx.Done(): + return + } + if err != nil { + return + } + } + }() + var cursor eventbus.Offset = eventbus.OffsetOldest + for i := range 30 { + select { + case err := <-committed: + if err != nil { + t.Fatal(err) + } + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } + events, next, err := reader.Read(ctx, cursor, 1) + if err != nil || len(events) != 1 || events[0].ID != fmt.Sprintf("concurrent-%d", i) { + t.Fatalf("WAL reader at %d: %v %v", i, events, err) + } + if order, err := reader.CompareOffsets(cursor, next); err != nil || order >= 0 { + t.Fatalf("opaque offsets did not advance: %q -> %q (%v)", cursor, next, err) + } + cursor = next + } + if _, err := os.Stat(path + "-wal"); err != nil { + t.Fatal("writer fixture did not retain a live WAL") + } + if err := writer.Close(); err != nil { + t.Fatal(err) + } + if events, tail, err := reader.Read(ctx, eventbus.OffsetNewest, 0); err != nil || len(events) != 0 || tail != cursor { + t.Fatalf("tail after writer close=%q err=%v", tail, err) + } +} diff --git a/stores/sqlite/store.go b/stores/sqlite/store.go index 4d3b78a..635ec8e 100644 --- a/stores/sqlite/store.go +++ b/stores/sqlite/store.go @@ -284,6 +284,12 @@ func (*SQLiteStore) CompareOffsets(left, right eventbus.Offset) (int, error) { // Append stores an event and returns its assigned offset. func (s *SQLiteStore) Append(ctx context.Context, event *eventbus.Event) (eventbus.Offset, error) { start := time.Now() + if s.cfg.readOnly { + if s.metricsHook != nil { + s.metricsHook.OnAppend(time.Since(start), ErrReadOnly) + } + return "", ErrReadOnly + } // Envelope fields are stored as NULL when absent so rows written through // old and new cores are indistinguishable on read (both scan as empty). @@ -457,6 +463,12 @@ func (s *SQLiteStore) scanEvents(rows rowScanner) ([]*eventbus.StoredEvent, erro // offset the store can resume from. func (s *SQLiteStore) SaveOffset(ctx context.Context, subscriptionID string, offset eventbus.Offset) error { start := time.Now() + if s.cfg.readOnly { + if s.metricsHook != nil { + s.metricsHook.OnSaveOffset(time.Since(start), ErrReadOnly) + } + return ErrReadOnly + } if offset == eventbus.OffsetNewest { position, err := s.resolveNewest(ctx) @@ -525,6 +537,9 @@ func (s *SQLiteStore) LoadOffset(ctx context.Context, subscriptionID string) (ev // SaveSnapshot upserts the compaction snapshot for snapshotID, recording that // blob reflects the projection as of (and including) atOffset. func (s *SQLiteStore) SaveSnapshot(ctx context.Context, snapshotID string, atOffset eventbus.Offset, blob json.RawMessage) error { + if s.cfg.readOnly { + return ErrReadOnly + } position, err := parseOffset(atOffset) if err != nil { return fmt.Errorf("sqlite: invalid snapshot offset: %w", err) @@ -555,6 +570,9 @@ func (s *SQLiteStore) LoadSnapshot(ctx context.Context, snapshotID string) (even // table uses AUTOINCREMENT, so deleted positions are never reused; a snapshot // resumed via Replay(atOffset) (which reads position > atOffset) is unaffected. func (s *SQLiteStore) TruncateBefore(ctx context.Context, beforeOffset eventbus.Offset) (int64, error) { + if s.cfg.readOnly { + return 0, ErrReadOnly + } position, err := parseOffset(beforeOffset) if err != nil { return 0, fmt.Errorf("sqlite: invalid truncate offset: %w", err)