feat(vfs): add observability PRAGMAs and relative time parsing

cmd/litestream-vfs/main.go

@@ -5,6 +5,11 @@ package main
 
 // import C is necessary export to the c-archive .a file
 
+/*
+#include <stdint.h>
+typedef int64_t sqlite3_int64;
+typedef uint64_t sqlite3_uint64;
+*/
 import "C"
 
 import (
@@ -14,6 +19,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"unsafe"
 
 	"github.com/psanford/sqlite3vfs"
 
@@ -65,3 +71,72 @@ func LitestreamVFSRegister() {
 		log.Fatalf("failed to register litestream vfs: %s", err)
 	}
 }
+
+//export GoLitestreamRegisterConnection
+func GoLitestreamRegisterConnection(dbPtr unsafe.Pointer, fileID C.sqlite3_uint64) *C.char {
+	if err := litestream.RegisterVFSConnection(uintptr(dbPtr), uint64(fileID)); err != nil {
+		return C.CString(err.Error())
+	}
+	return nil
+}
+
+//export GoLitestreamUnregisterConnection
+func GoLitestreamUnregisterConnection(dbPtr unsafe.Pointer) *C.char {
+	litestream.UnregisterVFSConnection(uintptr(dbPtr))
+	return nil
+}
+
+//export GoLitestreamSetTime
+func GoLitestreamSetTime(dbPtr unsafe.Pointer, timestamp *C.char) *C.char {
+	if timestamp == nil {
+		return C.CString("timestamp required")
+	}
+	if err := litestream.SetVFSConnectionTime(uintptr(dbPtr), C.GoString(timestamp)); err != nil {
+		return C.CString(err.Error())
+	}
+	return nil
+}
+
+//export GoLitestreamResetTime
+func GoLitestreamResetTime(dbPtr unsafe.Pointer) *C.char {
+	if err := litestream.ResetVFSConnectionTime(uintptr(dbPtr)); err != nil {
+		return C.CString(err.Error())
+	}
+	return nil
+}
+
+//export GoLitestreamTime
+func GoLitestreamTime(dbPtr unsafe.Pointer, out **C.char) *C.char {
+	value, err := litestream.GetVFSConnectionTime(uintptr(dbPtr))
+	if err != nil {
+		return C.CString(err.Error())
+	}
+	if out != nil {
+		*out = C.CString(value)
+	}
+	return nil
+}
+
+//export GoLitestreamTxid
+func GoLitestreamTxid(dbPtr unsafe.Pointer, out **C.char) *C.char {
+	value, err := litestream.GetVFSConnectionTXID(uintptr(dbPtr))
+	if err != nil {
+		return C.CString(err.Error())
+	}
+	if out != nil {
+		*out = C.CString(value)
+	}
+	return nil
+}
+
+//export GoLitestreamLag
+func GoLitestreamLag(dbPtr unsafe.Pointer, out *C.sqlite3_int64) *C.char {
+	value, err := litestream.GetVFSConnectionLag(uintptr(dbPtr))
+	if err != nil {
+		return C.CString(err.Error())
+	}
+	if out != nil {
+		*out = C.sqlite3_int64(value)
+	}
+	return nil
+}

cmd/litestream-vfs/time_travel_test.go

@@ -102,8 +102,188 @@ func TestVFS_TimeTravelFunctions(t *testing.T) {
 
 	if err := sqldb1.QueryRow("PRAGMA litestream_time").Scan(&currentTime); err != nil {
 		t.Fatalf("current time after reset: %v", err)
-	} else if currentTime != "latest" {
-		t.Fatalf("current time after reset mismatch: got %s, want latest", currentTime)
+	}
+	// After reset, should return actual LTX timestamp (not "latest" anymore per #853)
+	if _, err := time.Parse(time.RFC3339Nano, currentTime); err != nil {
+		t.Fatalf("current time after reset should be valid RFC3339Nano timestamp, got %s: %v", currentTime, err)
+	}
+}
+
+func TestVFS_PragmaLitestreamTxid(t *testing.T) {
+	ctx := context.Background()
+	client := file.NewReplicaClient(t.TempDir())
+	vfs := newVFS(t, client)
+	vfs.PollInterval = 50 * time.Millisecond
+	if err := sqlite3vfs.RegisterVFS("litestream-txid", vfs); err != nil {
+		t.Fatalf("failed to register litestream vfs: %v", err)
+	}
+
+	db := testingutil.NewDB(t, filepath.Join(t.TempDir(), "db"))
+	db.MonitorInterval = 50 * time.Millisecond
+	db.Replica = litestream.NewReplica(db)
+	db.Replica.Client = client
+	db.Replica.SyncInterval = 50 * time.Millisecond
+	if err := db.Open(); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = db.Close(ctx) }()
+
+	sqldb0 := testingutil.MustOpenSQLDB(t, db.Path())
+	defer testingutil.MustCloseSQLDB(t, sqldb0)
+
+	if _, err := sqldb0.Exec("CREATE TABLE t (x INTEGER)"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := sqldb0.Exec("INSERT INTO t (x) VALUES (100)"); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(6 * db.MonitorInterval)
+
+	sqldb1, err := sql.Open("sqlite3", "file:/tmp/txid-test.db?vfs=litestream-txid")
+	if err != nil {
+		t.Fatalf("failed to open database: %v", err)
+	}
+	defer sqldb1.Close()
+	sqldb1.SetMaxOpenConns(1)
+	time.Sleep(2 * vfs.PollInterval)
+
+	var txid int64
+	if err := sqldb1.QueryRow("PRAGMA litestream_txid").Scan(&txid); err != nil {
+		t.Fatalf("query txid: %v", err)
+	}
+	if txid <= 0 {
+		t.Fatalf("expected positive TXID, got %d", txid)
+	}
+
+	// Test that setting litestream_txid fails (read-only)
+	if _, err := sqldb1.Exec("PRAGMA litestream_txid = 123"); err == nil {
+		t.Fatal("expected error setting litestream_txid (read-only)")
+	}
+}
+
+func TestVFS_PragmaLitestreamLag(t *testing.T) {
+	ctx := context.Background()
+	client := file.NewReplicaClient(t.TempDir())
+	vfs := newVFS(t, client)
+	vfs.PollInterval = 50 * time.Millisecond
+	if err := sqlite3vfs.RegisterVFS("litestream-lag", vfs); err != nil {
+		t.Fatalf("failed to register litestream vfs: %v", err)
+	}
+
+	db := testingutil.NewDB(t, filepath.Join(t.TempDir(), "db"))
+	db.MonitorInterval = 50 * time.Millisecond
+	db.Replica = litestream.NewReplica(db)
+	db.Replica.Client = client
+	db.Replica.SyncInterval = 50 * time.Millisecond
+	if err := db.Open(); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = db.Close(ctx) }()
+
+	sqldb0 := testingutil.MustOpenSQLDB(t, db.Path())
+	defer testingutil.MustCloseSQLDB(t, sqldb0)
+
+	if _, err := sqldb0.Exec("CREATE TABLE t (x INTEGER)"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := sqldb0.Exec("INSERT INTO t (x) VALUES (100)"); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(6 * db.MonitorInterval)
+
+	sqldb1, err := sql.Open("sqlite3", "file:/tmp/lag-test.db?vfs=litestream-lag")
+	if err != nil {
+		t.Fatalf("failed to open database: %v", err)
+	}
+	defer sqldb1.Close()
+	sqldb1.SetMaxOpenConns(1)
+
+	// Wait for at least one successful poll (poll runs on ticker, not immediately)
+	time.Sleep(5 * vfs.PollInterval)
+
+	var lag int64
+	if err := sqldb1.QueryRow("PRAGMA litestream_lag").Scan(&lag); err != nil {
+		t.Fatalf("query lag: %v", err)
+	}
+	// Lag should be -1 (never polled) or a small non-negative value after polling
+	// -1 is valid if no poll has succeeded yet
+	if lag < -1 || lag > 5 {
+		t.Fatalf("expected lag between -1 and 5 seconds, got %d", lag)
+	}
+
+	// Test that setting litestream_lag fails (read-only)
+	if _, err := sqldb1.Exec("PRAGMA litestream_lag = 123"); err == nil {
+		t.Fatal("expected error setting litestream_lag (read-only)")
+	}
+}
+
+func TestVFS_PragmaRelativeTime(t *testing.T) {
+	ctx := context.Background()
+	client := file.NewReplicaClient(t.TempDir())
+	vfs := newVFS(t, client)
+	vfs.PollInterval = 50 * time.Millisecond
+	if err := sqlite3vfs.RegisterVFS("litestream-relative", vfs); err != nil {
+		t.Fatalf("failed to register litestream vfs: %v", err)
+	}
+
+	db := testingutil.NewDB(t, filepath.Join(t.TempDir(), "db"))
+	db.MonitorInterval = 50 * time.Millisecond
+	db.Replica = litestream.NewReplica(db)
+	db.Replica.Client = client
+	db.Replica.SyncInterval = 50 * time.Millisecond
+	if err := db.Open(); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = db.Close(ctx) }()
+
+	sqldb0 := testingutil.MustOpenSQLDB(t, db.Path())
+	defer testingutil.MustCloseSQLDB(t, sqldb0)
+
+	if _, err := sqldb0.Exec("CREATE TABLE t (x INTEGER)"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := sqldb0.Exec("INSERT INTO t (x) VALUES (100)"); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(6 * db.MonitorInterval)
+
+	sqldb1, err := sql.Open("sqlite3", "file:/tmp/relative-test.db?vfs=litestream-relative")
+	if err != nil {
+		t.Fatalf("failed to open database: %v", err)
+	}
+	defer sqldb1.Close()
+	sqldb1.SetMaxOpenConns(1)
+	time.Sleep(2 * vfs.PollInterval)
+
+	// Test that relative time parsing works (even if no data exists at that time)
+	// The parse should succeed, but may return "no backup files available" if too far in past
+	now := time.Now()
+	_, err = sqldb1.Exec("PRAGMA litestream_time = '1 second ago'")
+	// This might fail if no LTX files exist at that time, which is expected
+	// The important thing is that the parsing worked (not a "parse timestamp" error)
+	if err != nil && err.Error() != "no backup files available" {
+		// Check if it's a parse error vs a "no files" error
+		if err.Error() == "invalid timestamp (expected RFC3339 or relative time like '5 minutes ago'): 1 second ago" {
+			t.Fatalf("relative time parsing failed: %v", err)
+		}
+	}
+
+	// Reset to latest
+	if _, err := sqldb1.Exec("PRAGMA litestream_time = LATEST"); err != nil {
+		t.Fatalf("reset to latest: %v", err)
+	}
+
+	// Verify the current time is recent (within last minute)
+	var currentTime string
+	if err := sqldb1.QueryRow("PRAGMA litestream_time").Scan(&currentTime); err != nil {
+		t.Fatalf("query current time: %v", err)
+	}
+	ts, err := time.Parse(time.RFC3339Nano, currentTime)
+	if err != nil {
+		t.Fatalf("parse current time: %v", err)
+	}
+	if now.Sub(ts) > time.Minute {
+		t.Fatalf("current time too old: %v (now: %v)", ts, now)
 	}
 }
 

go.mod

@@ -86,9 +86,14 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
 	github.com/googleapis/gax-go/v2 v2.12.0 // indirect
+	github.com/hablullah/go-hijri v1.0.2 // indirect
+	github.com/hablullah/go-juliandays v1.0.0 // indirect
+	github.com/jalaali/go-jalaali v0.0.0-20210801064154-80525e88d958 // indirect
 	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/kr/fs v0.1.0 // indirect
 	github.com/kylelemons/godebug v1.1.0 // indirect
+	github.com/magefile/mage v1.14.0 // indirect
+	github.com/markusmobius/go-dateparser v1.2.4 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect
@@ -100,6 +105,8 @@ require (
 	github.com/prometheus/common v0.45.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/spf13/cast v1.7.1 // indirect
+	github.com/tetratelabs/wazero v1.2.1 // indirect
+	github.com/wasilibs/go-re2 v1.3.0 // indirect
 	github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1 // indirect

go.sum

@@ -159,10 +159,16 @@ github.com/gorilla/handlers v1.5.1 h1:9lRY6j8DEeeBT10CvO9hGW0gmky0BprnvDI5vfhUHH
 github.com/gorilla/handlers v1.5.1/go.mod h1:t8XrUpc4KVXb7HGyJ4/cEnwQiaxrX/hz1Zv/4g96P1Q=
 github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
+github.com/hablullah/go-hijri v1.0.2 h1:drT/MZpSZJQXo7jftf5fthArShcaMtsal0Zf/dnmp6k=
+github.com/hablullah/go-hijri v1.0.2/go.mod h1:OS5qyYLDjORXzK4O1adFw9Q5WfhOcMdAKglDkcTxgWQ=
+github.com/hablullah/go-juliandays v1.0.0 h1:A8YM7wIj16SzlKT0SRJc9CD29iiaUzpBLzh5hr0/5p0=
+github.com/hablullah/go-juliandays v1.0.0/go.mod h1:0JOYq4oFOuDja+oospuc61YoX+uNEn7Z6uHYTbBzdGc=
 github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
 github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/hokaccha/go-prettyjson v0.0.0-20211117102719-0474bc63780f h1:7LYC+Yfkj3CTRcShK0KOL/w6iTiKyqqBA9a41Wnggw8=
 github.com/hokaccha/go-prettyjson v0.0.0-20211117102719-0474bc63780f/go.mod h1:pFlLw2CfqZiIBOx6BuCeRLCrfxBJipTY0nIOF/VbGcI=
+github.com/jalaali/go-jalaali v0.0.0-20210801064154-80525e88d958 h1:qxLoi6CAcXVzjfvu+KXIXJOAsQB62LXjsfbOaErsVzE=
+github.com/jalaali/go-jalaali v0.0.0-20210801064154-80525e88d958/go.mod h1:Wqfu7mjUHj9WDzSSPI5KfBclTTEnLveRUFr/ujWnTgE=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/keybase/go-keychain v0.0.1 h1:way+bWYa6lDppZoZcgMbYsvC7GxljxrskdNInRtuthU=
@@ -179,8 +185,12 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
+github.com/magefile/mage v1.14.0 h1:6QDX3g6z1YvJ4olPhT1wksUcSa/V0a1B+pJb73fBjyo=
+github.com/magefile/mage v1.14.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A=
 github.com/mark3labs/mcp-go v0.32.0 h1:fgwmbfL2gbd67obg57OfV2Dnrhs1HtSdlY/i5fn7MU8=
 github.com/mark3labs/mcp-go v0.32.0/go.mod h1:rXqOudj/djTORU/ThxYx8fqEVj/5pvTuuebQ2RC7uk4=
+github.com/markusmobius/go-dateparser v1.2.4 h1:2e8XJozaERVxGwsRg72coi51L2aiYqE2gukkdLc85ck=
+github.com/markusmobius/go-dateparser v1.2.4/go.mod h1:CBAUADJuMNhJpyM6IYaWAoFhtKaqnUcznY2cL7gNugY=
 github.com/mattn/go-colorable v0.1.9/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
 github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
 github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
@@ -256,6 +266,10 @@ github.com/studio-b12/gowebdav v0.11.0 h1:qbQzq4USxY28ZYsGJUfO5jR+xkFtcnwWgitp4Z
 github.com/studio-b12/gowebdav v0.11.0/go.mod h1:bHA7t77X/QFExdeAnDzK6vKM34kEZAcE1OX4MfiwjkE=
 github.com/superfly/ltx v0.5.0 h1:dXNrcT3ZtMb6iKZopIV7z5UBscnapg0b0F02loQsk5o=
 github.com/superfly/ltx v0.5.0/go.mod h1:Nf50QAIXU/ET4ua3AuQ2fh31MbgNQZA7r/DYx6Os77s=
+github.com/tetratelabs/wazero v1.2.1 h1:J4X2hrGzJvt+wqltuvcSjHQ7ujQxA9gb6PeMs4qlUWs=
+github.com/tetratelabs/wazero v1.2.1/go.mod h1:wYx2gNRg8/WihJfSDxA1TIL8H+GkfLYm+bIfbblu9VQ=
+github.com/wasilibs/go-re2 v1.3.0 h1:LFhBNzoStM3wMie6rN2slD1cuYH2CGiHpvNL3UtcsMw=
+github.com/wasilibs/go-re2 v1.3.0/go.mod h1:AafrCXVvGRJJOImMajgJ2M7rVmWyisVK7sFshbxnVrg=
 github.com/yosida95/uritemplate/v3 v3.0.2 h1:Ed3Oyj9yrmi9087+NczuL5BwkIc4wvTb5zIM+UJPGz4=
 github.com/yosida95/uritemplate/v3 v3.0.2/go.mod h1:ILOh0sOhIJR3+L/8afwt/kE++YT040gmv5BQTMR2HP4=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=