diff --git a/errors.go b/errors.go new file mode 100644 index 0000000..58e910b --- /dev/null +++ b/errors.go @@ -0,0 +1,15 @@ +package sqlitestream + +import "errors" + +// ErrNotReplicating is returned when an operation requires a configured +// replica but replication is off. +var ErrNotReplicating = errors.New("sqlitestream: replication not configured") + +// ErrExists is returned by Restore when the output path already exists +// and force is false. +var ErrExists = errors.New("sqlitestream: file exists") + +// ErrReplicaEmpty is returned by Restore when the replica holds no +// snapshots. Callers can treat this as a fresh install. +var ErrReplicaEmpty = errors.New("sqlitestream: replica is empty — nothing to restore") diff --git a/restore.go b/restore.go index da7b889..948f738 100644 --- a/restore.go +++ b/restore.go @@ -17,13 +17,13 @@ import ( func Restore(ctx context.Context, outputPath string, force bool, opts ...Option) error { cfg := apply(opts) if cfg.replicaURL == "" { - return errors.New("WithReplicaURL is required for restore") + return fmt.Errorf("%w: WithReplicaURL is required for restore", ErrNotReplicating) } switch _, err := os.Stat(outputPath); { case err == nil: if !force { - return fmt.Errorf("%s exists (pass force=true to overwrite)", outputPath) + return fmt.Errorf("%w: %s (pass force=true to overwrite)", ErrExists, outputPath) } if err := removeDBFiles(outputPath); err != nil { return fmt.Errorf("clear destination: %w", err) @@ -31,7 +31,7 @@ func Restore(ctx context.Context, outputPath string, force bool, opts ...Option) case errors.Is(err, os.ErrNotExist): // ok — nothing to clear default: - return fmt.Errorf("unexpected error openning path: %s (%w)", outputPath, err) + return fmt.Errorf("unexpected error opening path: %s (%w)", outputPath, err) } db, err := newLitestreamDB(outputPath, cfg) @@ -41,8 +41,10 @@ func Restore(ctx context.Context, outputPath string, force bool, opts ...Option) opt := litestream.NewRestoreOptions() opt.OutputPath = outputPath if err := db.Replica.Restore(ctx, opt); err != nil { - if errors.Is(err, litestream.ErrNoSnapshots) { - return errors.New("replica is empty — nothing to restore") + // litestream reports an empty replica as either error, depending + // on the client. + if errors.Is(err, litestream.ErrNoSnapshots) || errors.Is(err, litestream.ErrTxNotAvailable) { + return ErrReplicaEmpty } return fmt.Errorf("restore: %w", err) } diff --git a/sqlitestream.go b/sqlitestream.go index e34f08d..0f26e95 100644 --- a/sqlitestream.go +++ b/sqlitestream.go @@ -6,6 +6,7 @@ import ( "fmt" "log/slog" "net/url" + "time" "github.com/benbjohnson/litestream" "github.com/benbjohnson/litestream/s3" // also registers the s3:// scheme @@ -64,6 +65,37 @@ func (d *DB) DB() *sql.DB { return d.sql } +// Replicating reports whether background replication is configured. +func (d *DB) Replicating() bool { + return d != nil && d.ldb != nil +} + +// Retention reports whether litestream is handling retention. +func (d *DB) Retention() bool { + return d != nil && d.ldb != nil && d.ldb.RetentionEnabled +} + +// LastSyncedAt returns the time of the last successful sync to the +// replica. The zero time means replication is off or nothing has been +// synced yet. +func (d *DB) LastSyncedAt() time.Time { + if d == nil || d.ldb == nil { + return time.Time{} + } + return d.ldb.LastSuccessfulSyncAt() +} + +// SyncStatus reports the status between the local database and the +// configured (cloud) storage by comparing transaction records. This +// may entail I/O. Returns ErrNotReplicating when no replica is +// configured. +func (d *DB) SyncStatus(ctx context.Context) (litestream.SyncStatus, error) { + if d == nil || d.ldb == nil { + return litestream.SyncStatus{}, ErrNotReplicating + } + return d.ldb.SyncStatus(ctx) +} + // Close flushes any pending WAL to S3 (when replication is on), stops the // replicator, and closes the SQL handle. // diff --git a/sqlitestream_test.go b/sqlitestream_test.go index 88ba192..fa65480 100644 --- a/sqlitestream_test.go +++ b/sqlitestream_test.go @@ -4,7 +4,9 @@ import ( "os" "path/filepath" "testing" + "time" + _ "github.com/benbjohnson/litestream/file" // registers the file:// scheme for tests "github.com/luzilla/sqlitestream" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -41,10 +43,53 @@ func TestSuite(t *testing.T) { assert.NoError(t, db.Close(t.Context())) }) + // Sync status accessors are safe on a nil receiver and without a replica. + t.Run("SyncStatusNoReplica", func(t *testing.T) { + var nilDB *sqlitestream.DB + assert.False(t, nilDB.Replicating()) + assert.False(t, nilDB.Retention()) + assert.True(t, nilDB.LastSyncedAt().IsZero()) + _, err := nilDB.SyncStatus(t.Context()) + assert.ErrorIs(t, err, sqlitestream.ErrNotReplicating) + + db, err := sqlitestream.Open(t.Context(), filepath.Join(t.TempDir(), "test.db")) + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close(t.Context()) }) + + assert.False(t, db.Replicating()) + assert.False(t, db.Retention()) + assert.True(t, db.LastSyncedAt().IsZero()) + _, err = db.SyncStatus(t.Context()) + assert.ErrorIs(t, err, sqlitestream.ErrNotReplicating) + }) + + // With a replica configured, LastSyncedAt reflects the background sync. + t.Run("SyncStatusWithReplica", func(t *testing.T) { + replicaURL := "file://" + filepath.Join(t.TempDir(), "replica") + db, err := sqlitestream.Open(t.Context(), filepath.Join(t.TempDir(), "test.db"), + sqlitestream.WithReplicaURL(replicaURL)) + require.NoError(t, err) + t.Cleanup(func() { _ = db.Close(t.Context()) }) + + assert.True(t, db.Replicating()) + assert.True(t, db.Retention()) // litestream default + + _, err = db.DB().Exec(`CREATE TABLE t (id INTEGER PRIMARY KEY, v TEXT)`) + require.NoError(t, err) + + require.Eventually(t, func() bool { + return !db.LastSyncedAt().IsZero() + }, 10*time.Second, 100*time.Millisecond, "no sync recorded") + assert.WithinDuration(t, time.Now(), db.LastSyncedAt(), time.Minute) + + _, err = db.SyncStatus(t.Context()) + assert.NoError(t, err) + }) + t.Run("restore", func(t *testing.T) { t.Run("RequiresReplicaURL", func(t *testing.T) { err := sqlitestream.Restore(t.Context(), filepath.Join(t.TempDir(), "out.db"), false) - require.Error(t, err) + require.ErrorIs(t, err, sqlitestream.ErrNotReplicating) assert.Contains(t, err.Error(), "WithReplicaURL") }) @@ -56,9 +101,13 @@ func TestSuite(t *testing.T) { err := sqlitestream.Restore(t.Context(), path, false, sqlitestream.WithReplicaURL("file:///nonexistent")) - require.Error(t, err) - assert.Contains(t, err.Error(), "exists") + require.ErrorIs(t, err, sqlitestream.ErrExists) }) + t.Run("EmptyReplica", func(t *testing.T) { + err := sqlitestream.Restore(t.Context(), filepath.Join(t.TempDir(), "out.db"), false, + sqlitestream.WithReplicaURL("file://"+t.TempDir())) + require.ErrorIs(t, err, sqlitestream.ErrReplicaEmpty) + }) }) }