Repository: benbjohnson/litestream

Issue:
feat: Add native Alibaba Cloud OSS storage backend support
## Feature Description

Add native support for Alibaba Cloud Object Storage Service (OSS) as a storage backend, similar
to existing S3, GCS, and ABS implementations.

## Use Cases

I was using Litestream 0.3.x with Alibaba Cloud OSS through its S3-compatible API. However,
after upgrading to Litestream 0.5.x, this compatibility was broken.

**Root cause:** Litestream 0.5 upgraded to AWS SDK Go v2, which automatically enables chunked
encoding for uploads ≥5MB. Alibaba Cloud OSS does not support chunked encoding with
S3-compatible mode.

According to [Alibaba Cloud's official 
documentation](https://www.alibabacloud.com/help/en/oss/developer-reference/use-aws-sdks-to-access-oss#2842f2f03d5bw):
> Go SDK v2 automatically enables chunked encoding for large files (≥5MB), causing upload
failures. Use Go SDK v1 for large file uploads.

This makes it impossible for users in China and other regions who rely on Alibaba Cloud OSS to
use Litestream 0.5+.

## Proposal

I have implemented a native OSS backend using the official
[aliyun-oss-go-sdk](https://github.com/aliyun/aliyun-oss-go-sdk). The implementation:

- Follows the existing `ReplicaClient` interface pattern
- Supports all required operations (LTXFiles, OpenLTXFile, WriteLTXFile, DeleteLTXFiles, etc.)
- Handles metadata timestamps for point-in-time recovery
- Supports environment variable credentials (`ALIBABA_CLOUD_ACCESS_KEY_ID`,
`ALIBABA_CLOUD_ACCESS_KEY_SECRET`)
- Supports custom endpoints and regions

**Configuration example:**
```yaml
dbs:
  - path: /path/to/db.sqlite
    replicas:
      - type: oss
        bucket: my-bucket
        path: backups
        region: cn-hangzhou
        access-key-id: ${ALIBABA_CLOUD_ACCESS_KEY_ID}
        access-key-secret: ${ALIBABA_CLOUD_ACCESS_KEY_SECRET}

Or using URL format:
dbs:
  - path: /path/to/db.sqlite
    replicas:
      - url: oss://my-bucket/backups
```

I'm happy to submit a PR with the implementation and maintain the OSS backend going forward.

https://github.com/cosformula/litestream/pull/1

Hints from issue comments:

Solution patch:
diff --git a/AGENTS.md b/AGENTS.md
index 4eae325..6408dd8 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -23,7 +23,7 @@ Litestream is a **disaster recovery tool for SQLite** that runs as a background
 - **Multi-level Compaction**: Hierarchical compaction keeps storage efficient (30s → 5m → 1h → snapshots)
 - **Single Replica Constraint**: Each database is replicated to exactly one remote destination
 - **Pure Go Build**: Uses `modernc.org/sqlite`, so no CGO dependency for the main binary
-- **Optional NATS JetStream Support**: Additional replica backend alongside S3/GCS/ABS/File/SFTP
+- **Optional NATS JetStream Support**: Additional replica backend alongside S3/GCS/ABS/OSS/File/SFTP
 - **Snapshot Compatibility**: Only LTX-based backups are supported—keep legacy v0.3.x binaries to restore old WAL snapshots
 
 **Key Design Principles:**
@@ -728,7 +728,7 @@ if offset < 0 {
 
 ```bash
 # Test eventual consistency
-go test -v ./replica_client_test.go -integration [s3|gcs|abs|sftp]
+go test -v ./replica_client_test.go -integration [s3|gcs|abs|oss|sftp]
 
 # Test partial reads
 # (Example) add targeted partial-read tests in your backend package
diff --git a/CLAUDE.md b/CLAUDE.md
index 23fb48d..11b4191 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -86,6 +86,7 @@ CGO_ENABLED=1 go build -tags vfs -o bin/litestream-vfs ./cmd/litestream-vfs  # T
 go test -v ./replica_client_test.go -integration s3
 go test -v ./replica_client_test.go -integration gcs
 go test -v ./replica_client_test.go -integration abs
+go test -v ./replica_client_test.go -integration oss
 go test -v ./replica_client_test.go -integration sftp
 ```
 
@@ -112,7 +113,7 @@ pre-commit run --all-files
 
 **Replica (`replica.go`)**: Connects a database to replication destinations via ReplicaClient interface. Manages periodic synchronization and maintains replication position.
 
-**ReplicaClient Interface** (`replica_client.go`): Abstraction for different storage backends (S3, GCS, Azure Blob Storage, SFTP, file system, NATS). Each implementation handles snapshot/WAL segment upload and restoration. The `LTXFiles` method includes a `useMetadata` parameter: when true, it fetches accurate timestamps from backend metadata (required for point-in-time restores); when false, it uses fast timestamps for normal operations. During compaction, the system preserves the earliest CreatedAt timestamp from source files to maintain temporal granularity for restoration.
+**ReplicaClient Interface** (`replica_client.go`): Abstraction for different storage backends (S3, GCS, Azure Blob Storage, OSS, SFTP, file system, NATS). Each implementation handles snapshot/WAL segment upload and restoration. The `LTXFiles` method includes a `useMetadata` parameter: when true, it fetches accurate timestamps from backend metadata (required for point-in-time restores); when false, it uses fast timestamps for normal operations. During compaction, the system preserves the earliest CreatedAt timestamp from source files to maintain temporal granularity for restoration.
 
 **WAL Processing**: The system monitors SQLite WAL files for changes, segments them into LTX format files, and replicates these segments to configured destinations. Uses SQLite checksums for integrity verification.
 
@@ -121,6 +122,7 @@ pre-commit run --all-files
 - **S3** (`s3/replica_client.go`): AWS S3 and compatible storage
 - **GCS** (`gs/replica_client.go`): Google Cloud Storage
 - **ABS** (`abs/replica_client.go`): Azure Blob Storage
+- **OSS** (`oss/replica_client.go`): Alibaba Cloud Object Storage Service
 - **SFTP** (`sftp/replica_client.go`): SSH File Transfer Protocol
 - **File** (`file/replica_client.go`): Local file system replication
 - **NATS** (`nats/replica_client.go`): NATS JetStream object storage
diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md
index a00d281..802b7ce 100644
--- a/docs/ARCHITECTURE.md
+++ b/docs/ARCHITECTURE.md
@@ -37,6 +37,7 @@ graph TB
         S3[s3/replica_client.go]
         GCS[gs/replica_client.go]
         ABS[abs/replica_client.go]
+        OSS[oss/replica_client.go]
         File[file/replica_client.go]
         SFTP[sftp/replica_client.go]
         NATS[nats/replica_client.go]
@@ -54,6 +55,7 @@ graph TB
     RC --> S3
     RC --> GCS
     RC --> ABS
+    RC --> OSS
     RC --> File
     RC --> SFTP
     RC --> NATS
@@ -61,6 +63,7 @@ graph TB
     S3 --> Cloud
     GCS --> Cloud
     ABS --> Cloud
+    OSS --> Cloud
 ```
 
 ### Layer Responsibilities
diff --git a/go.mod b/go.mod
index 51c0a44..b7c05a5 100644
--- a/go.mod
+++ b/go.mod
@@ -52,6 +52,8 @@ require (
 	modernc.org/memory v1.11.0 // indirect
 )
 
+require github.com/aliyun/alibabacloud-oss-go-sdk-v2 v1.3.0
+
 require (
 	cloud.google.com/go v0.111.0 // indirect
 	cloud.google.com/go/compute v1.23.3 // indirect
@@ -86,9 +88,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 +107,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
diff --git a/go.sum b/go.sum
index 94a9f32..7cac3d2 100644
--- a/go.sum
+++ b/go.sum
@@ -36,6 +36,8 @@ github.com/MadAppGang/httplog v1.3.0 h1:1XU54TO8kiqTeO+7oZLKAM3RP/cJ7SadzslRcKsp
 github.com/MadAppGang/httplog v1.3.0/go.mod h1:gpYEdkjh/Cda6YxtDy4AB7KY+fR7mb3SqBZw74A5hJ4=
 github.com/TylerBrock/colorjson v0.0.0-20200706003622-8a50f05110d2 h1:ZBbLwSJqkHBuFDA6DUhhse0IGJ7T5bemHyNILUjvOq4=
 github.com/TylerBrock/colorjson v0.0.0-20200706003622-8a50f05110d2/go.mod h1:VSw57q4QFiWDbRnjdX8Cb3Ow0SFncRw+bA/ofY6Q83w=
+github.com/aliyun/alibabacloud-oss-go-sdk-v2 v1.3.0 h1:wQlqotpyjYPjJz+Noh5bRu7Snmydk8SKC5Z6u1CR20Y=
+github.com/aliyun/alibabacloud-oss-go-sdk-v2 v1.3.0/go.mod h1:FTzydeQVmR24FI0D6XWUOMKckjXehM/jgMn1xC+DA9M=
 github.com/aws/aws-sdk-go-v2 v1.37.1 h1:SMUxeNz3Z6nqGsXv0JuJXc8w5YMtrQMuIBmDx//bBDY=
 github.com/aws/aws-sdk-go-v2 v1.37.1/go.mod h1:9Q0OoGQoboYIAJyslFyF1f5K1Ryddop8gqMhWx/n4Wg=
 github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.0 h1:6GMWV6CNpA/6fbFHnoAjrv4+LGfyTqZz2LtCHnspgDg=
@@ -157,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=
@@ -177,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 +268,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=
diff --git a/oss/replica_client.go b/oss/replica_client.go
new file mode 100644
index 0000000..4881a0f
--- /dev/null
+++ b/oss/replica_client.go
@@ -0,0 +1,618 @@
+package oss
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/url"
+	"os"
+	"path"
+	"regexp"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
+	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
+	"github.com/superfly/ltx"
+
+	"github.com/benbjohnson/litestream"
+	"github.com/benbjohnson/litestream/internal"
+)
+
+// ReplicaClientType is the client type for this package.
+const ReplicaClientType = "oss"
+
+// MetadataKeyTimestamp is the metadata key for storing LTX file timestamps in OSS.
+// Note: OSS SDK automatically adds "x-oss-meta-" prefix when setting metadata.
+const MetadataKeyTimestamp = "litestream-timestamp"
+
+// MaxKeys is the number of keys OSS can operate on per batch.
+const MaxKeys = 1000
+
+// DefaultRegion is the region used if one is not specified.
+const DefaultRegion = "cn-hangzhou"
+
+var _ litestream.ReplicaClient = (*ReplicaClient)(nil)
+
+// ReplicaClient is a client for writing LTX files to Alibaba Cloud OSS.
+type ReplicaClient struct {
+	mu       sync.Mutex
+	client   *oss.Client
+	uploader *oss.Uploader
+	logger   *slog.Logger
+
+	// Alibaba Cloud authentication keys.
+	AccessKeyID     string
+	AccessKeySecret string
+
+	// OSS bucket information
+	Region   string
+	Bucket   string
+	Path     string
+	Endpoint string
+
+	// Upload configuration
+	PartSize    int64 // Part size for multipart uploads (default: 5MB)
+	Concurrency int   // Number of concurrent parts to upload (default: 3)
+}
+
+// NewReplicaClient returns a new instance of ReplicaClient.
+func NewReplicaClient() *ReplicaClient {
+	return &ReplicaClient{
+		logger: slog.Default().WithGroup(ReplicaClientType),
+	}
+}
+
+// Type returns "oss" as the client type.
+func (c *ReplicaClient) Type() string {
+	return ReplicaClientType
+}
+
+// Init initializes the connection to OSS. No-op if already initialized.
+func (c *ReplicaClient) Init(ctx context.Context) (err error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.client != nil {
+		return nil
+	}
+
+	// Validate required configuration
+	if c.Bucket == "" {
+		return fmt.Errorf("oss: bucket name is required")
+	}
+
+	// Use default region if not specified
+	region := c.Region
+	if region == "" {
+		region = DefaultRegion
+	}
+
+	// Build configuration
+	cfg := oss.LoadDefaultConfig()
+
+	// Configure credentials
+	if c.AccessKeyID != "" && c.AccessKeySecret != "" {
+		cfg = cfg.WithCredentialsProvider(
+			credentials.NewStaticCredentialsProvider(c.AccessKeyID, c.AccessKeySecret),
+		)
+	} else {
+		// Use environment variable credentials provider
+		cfg = cfg.WithCredentialsProvider(
+			credentials.NewEnvironmentVariableCredentialsProvider(),
+		)
+	}
+
+	// Configure region
+	cfg = cfg.WithRegion(region)
+
+	// Configure custom endpoint if specified
+	if c.Endpoint != "" {
+		endpoint := c.Endpoint
+		// Add scheme if not present
+		if !strings.HasPrefix(endpoint, "http://") && !strings.HasPrefix(endpoint, "https://") {
+			endpoint = "https://" + endpoint
+		}
+		cfg = cfg.WithEndpoint(endpoint)
+	}
+
+	// Create OSS client
+	c.client = oss.NewClient(cfg)
+
+	// Create uploader with configurable part size and concurrency
+	uploaderOpts := []func(*oss.UploaderOptions){}
+	if c.PartSize > 0 {
+		uploaderOpts = append(uploaderOpts, func(o *oss.UploaderOptions) {
+			o.PartSize = c.PartSize
+		})
+	}
+	if c.Concurrency > 0 {
+		uploaderOpts = append(uploaderOpts, func(o *oss.UploaderOptions) {
+			o.ParallelNum = c.Concurrency
+		})
+	}
+	c.uploader = c.client.NewUploader(uploaderOpts...)
+
+	return nil
+}
+
+// LTXFiles returns an iterator over all LTX files on the replica for the given level.
+// When useMetadata is true, fetches accurate timestamps from OSS metadata via HeadObject.
+// When false, uses fast LastModified timestamps from LIST operation.
+func (c *ReplicaClient) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) {
+	if err := c.Init(ctx); err != nil {
+		return nil, err
+	}
+	return newFileIterator(ctx, c, level, seek, useMetadata), nil
+}
+
+// OpenLTXFile returns a reader for an LTX file.
+// Returns os.ErrNotExist if no matching file is found.
+func (c *ReplicaClient) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
+	if err := c.Init(ctx); err != nil {
+		return nil, err
+	}
+
+	// Build the key from the file info
+	filename := ltx.FormatFilename(minTXID, maxTXID)
+	key := c.ltxPath(level, filename)
+
+	request := &oss.GetObjectRequest{
+		Bucket: oss.Ptr(c.Bucket),
+		Key:    oss.Ptr(key),
+	}
+
+	// Set range header if offset is specified
+	if size > 0 {
+		request.RangeBehavior = oss.Ptr("standard")
+		request.Range = oss.Ptr(fmt.Sprintf("bytes=%d-%d", offset, offset+size-1))
+	} else if offset > 0 {
+		request.RangeBehavior = oss.Ptr("standard")
+		request.Range = oss.Ptr(fmt.Sprintf("bytes=%d-", offset))
+	}
+
+	result, err := c.client.GetObject(ctx, request)
+	if err != nil {
+		if isNotExists(err) {
+			return nil, os.ErrNotExist
+		}
+		return nil, fmt.Errorf("oss: get object %s: %w", key, err)
+	}
+
+	internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "GET").Inc()
+
+	return result.Body, nil
+}
+
+// WriteLTXFile writes an LTX file to the replica.
+// Extracts timestamp from LTX header and stores it in OSS metadata to preserve original creation time.
+// Uses multipart upload for large files via the uploader.
+func (c *ReplicaClient) WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
+	if err := c.Init(ctx); err != nil {
+		return nil, err
+	}
+
+	// Use TeeReader to peek at LTX header while preserving data for upload
+	var buf bytes.Buffer
+	teeReader := io.TeeReader(r, &buf)
+
+	// Extract timestamp from LTX header
+	hdr, _, err := ltx.PeekHeader(teeReader)
+	if err != nil {
+		return nil, fmt.Errorf("extract timestamp from LTX header: %w", err)
+	}
+	timestamp := time.UnixMilli(hdr.Timestamp).UTC()
+
+	// Combine buffered data with rest of reader
+	rc := internal.NewReadCounter(io.MultiReader(&buf, r))
+
+	filename := ltx.FormatFilename(minTXID, maxTXID)
+	key := c.ltxPath(level, filename)
+
+	// Store timestamp in OSS metadata for accurate timestamp retrieval
+	metadata := map[string]string{
+		MetadataKeyTimestamp: timestamp.Format(time.RFC3339Nano),
+	}
+
+	// Use uploader for automatic multipart handling (files >5GB)
+	result, err := c.uploader.UploadFrom(ctx, &oss.PutObjectRequest{
+		Bucket:   oss.Ptr(c.Bucket),
+		Key:      oss.Ptr(key),
+		Metadata: metadata,
+	}, rc)
+	if err != nil {
+		return nil, fmt.Errorf("oss: upload to %s: %w", key, err)
+	}
+
+	// Build file info from the uploaded file
+	info := &ltx.FileInfo{
+		Level:     level,
+		MinTXID:   minTXID,
+		MaxTXID:   maxTXID,
+		Size:      rc.N(),
+		CreatedAt: timestamp,
+	}
+
+	internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "PUT").Inc()
+	internal.OperationBytesCounterVec.WithLabelValues(ReplicaClientType, "PUT").Add(float64(rc.N()))
+
+	// ETag indicates successful upload
+	if result.ETag == nil || *result.ETag == "" {
+		return nil, fmt.Errorf("oss: upload failed: no ETag returned")
+	}
+
+	return info, nil
+}
+
+// DeleteLTXFiles deletes one or more LTX files.
+func (c *ReplicaClient) DeleteLTXFiles(ctx context.Context, a []*ltx.FileInfo) error {
+	if err := c.Init(ctx); err != nil {
+		return err
+	}
+
+	if len(a) == 0 {
+		return nil
+	}
+
+	// Convert file infos to object identifiers
+	objects := make([]oss.DeleteObject, 0, len(a))
+	for _, info := range a {
+		filename := ltx.FormatFilename(info.MinTXID, info.MaxTXID)
+		key := c.ltxPath(info.Level, filename)
+		objects = append(objects, oss.DeleteObject{Key: oss.Ptr(key)})
+
+		c.logger.Debug("deleting ltx file", "level", info.Level, "minTXID", info.MinTXID, "maxTXID", info.MaxTXID, "key", key)
+	}
+
+	// Delete in batches
+	for len(objects) > 0 {
+		n := min(len(objects), MaxKeys)
+		batch := objects[:n]
+
+		request := &oss.DeleteMultipleObjectsRequest{
+			Bucket:  oss.Ptr(c.Bucket),
+			Objects: batch,
+		}
+
+		out, err := c.client.DeleteMultipleObjects(ctx, request)
+		if err != nil {
+			return fmt.Errorf("oss: delete batch of %d objects: %w", n, err)
+		} else if err := deleteResultError(batch, out); err != nil {
+			return err
+		}
+
+		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "DELETE").Inc()
+
+		objects = objects[n:]
+	}
+
+	return nil
+}
+
+// DeleteAll deletes all files.
+func (c *ReplicaClient) DeleteAll(ctx context.Context) error {
+	if err := c.Init(ctx); err != nil {
+		return err
+	}
+
+	var objects []oss.DeleteObject
+
+	// Create paginator for listing objects
+	prefix := c.Path + "/"
+	paginator := c.client.NewListObjectsV2Paginator(&oss.ListObjectsV2Request{
+		Bucket: oss.Ptr(c.Bucket),
+		Prefix: oss.Ptr(prefix),
+	})
+
+	// Iterate through all pages
+	for paginator.HasNext() {
+		page, err := paginator.NextPage(ctx)
+		if err != nil {
+			return fmt.Errorf("oss: list objects page: %w", err)
+		}
+
+		// Collect object identifiers
+		for _, obj := range page.Contents {
+			if obj.Key != nil {
+				objects = append(objects, oss.DeleteObject{Key: obj.Key})
+			}
+		}
+	}
+
+	// Delete all collected objects in batches
+	for len(objects) > 0 {
+		n := min(len(objects), MaxKeys)
+		batch := objects[:n]
+
+		request := &oss.DeleteMultipleObjectsRequest{
+			Bucket:  oss.Ptr(c.Bucket),
+			Objects: batch,
+		}
+
+		out, err := c.client.DeleteMultipleObjects(ctx, request)
+		if err != nil {
+			return fmt.Errorf("oss: delete all batch of %d objects: %w", n, err)
+		} else if err := deleteResultError(batch, out); err != nil {
+			return err
+		}
+
+		objects = objects[n:]
+	}
+
+	return nil
+}
+
+// ltxPath returns the full path to an LTX file.
+func (c *ReplicaClient) ltxPath(level int, filename string) string {
+	return c.Path + "/" + fmt.Sprintf("%04x/%s", level, filename)
+}
+
+// fileIterator represents an iterator over LTX files in OSS.
+type fileIterator struct {
+	ctx         context.Context
+	cancel      context.CancelFunc
+	client      *ReplicaClient
+	level       int
+	seek        ltx.TXID
+	useMetadata bool // When true, fetch accurate timestamps from metadata
+
+	paginator   *oss.ListObjectsV2Paginator
+	page        *oss.ListObjectsV2Result
+	pageIndex   int
+	initialized bool
+
+	closed bool
+	err    error
+	info   *ltx.FileInfo
+}
+
+func newFileIterator(ctx context.Context, client *ReplicaClient, level int, seek ltx.TXID, useMetadata bool) *fileIterator {
+	ctx, cancel := context.WithCancel(ctx)
+
+	itr := &fileIterator{
+		ctx:         ctx,
+		cancel:      cancel,
+		client:      client,
+		level:       level,
+		seek:        seek,
+		useMetadata: useMetadata,
+	}
+
+	return itr
+}
+
+// initPaginator initializes the paginator lazily.
+func (itr *fileIterator) initPaginator() {
+	if itr.initialized {
+		return
+	}
+	itr.initialized = true
+
+	// Create paginator for listing objects with level prefix
+	prefix := itr.client.ltxPath(itr.level, "")
+	itr.paginator = itr.client.client.NewListObjectsV2Paginator(&oss.ListObjectsV2Request{
+		Bucket: oss.Ptr(itr.client.Bucket),
+		Prefix: oss.Ptr(prefix),
+	})
+}
+
+// Close stops iteration.
+func (itr *fileIterator) Close() (err error) {
+	itr.closed = true
+	itr.cancel()
+	return nil
+}
+
+// Next returns the next file. Returns false when no more files are available.
+func (itr *fileIterator) Next() bool {
+	if itr.closed || itr.err != nil {
+		return false
+	}
+
+	// Initialize paginator on first call
+	itr.initPaginator()
+
+	// Process objects until we find a valid LTX file
+	for {
+		// Load next page if needed
+		if itr.page == nil || itr.pageIndex >= len(itr.page.Contents) {
+			if !itr.paginator.HasNext() {
+				return false
+			}
+
+			var err error
+			itr.page, err = itr.paginator.NextPage(itr.ctx)
+			if err != nil {
+				itr.err = err
+				return false
+			}
+			itr.pageIndex = 0
+		}
+
+		// Process current object
+		if itr.pageIndex < len(itr.page.Contents) {
+			obj := itr.page.Contents[itr.pageIndex]
+			itr.pageIndex++
+
+			if obj.Key == nil {
+				continue
+			}
+
+			// Extract file info from key
+			key := path.Base(*obj.Key)
+			minTXID, maxTXID, err := ltx.ParseFilename(key)
+			if err != nil {
+				continue // Skip non-LTX files
+			}
+
+			// Build file info
+			info := &ltx.FileInfo{
+				Level:   itr.level,
+				MinTXID: minTXID,
+				MaxTXID: maxTXID,
+			}
+
+			// Skip if below seek TXID
+			if info.MinTXID < itr.seek {
+				continue
+			}
+
+			// Set file info
+			info.Size = obj.Size
+
+			// Use fast LastModified timestamp by default
+			var createdAt time.Time
+			if obj.LastModified != nil {
+				createdAt = obj.LastModified.UTC()
+			} else {
+				createdAt = time.Now().UTC()
+			}
+
+			// Only fetch accurate timestamp from metadata when requested (timestamp-based restore)
+			if itr.useMetadata {
+				head, err := itr.client.client.HeadObject(itr.ctx, &oss.HeadObjectRequest{
+					Bucket: oss.Ptr(itr.client.Bucket),
+					Key:    obj.Key,
+				})
+				if err != nil {
+					itr.err = fmt.Errorf("fetch object metadata: %w", err)
+					return false
+				}
+
+				if head.Metadata != nil {
+					if ts, ok := head.Metadata[MetadataKeyTimestamp]; ok {
+						if parsed, err := time.Parse(time.RFC3339Nano, ts); err == nil {
+							createdAt = parsed
+						} else {
+							itr.err = fmt.Errorf("parse timestamp from metadata: %w", err)
+							return false
+						}
+					}
+				}
+			}
+
+			info.CreatedAt = createdAt
+			itr.info = info
+			return true
+		}
+	}
+}
+
+// Item returns the metadata for the current file.
+func (itr *fileIterator) Item() *ltx.FileInfo {
+	return itr.info
+}
+
+// Err returns any error that occurred during iteration.
+func (itr *fileIterator) Err() error {
+	return itr.err
+}
+
+// ParseURL parses an OSS URL into its host and path parts.
+func ParseURL(s string) (bucket, region, key string, err error) {
+	u, err := url.Parse(s)
+	if err != nil {
+		return "", "", "", err
+	}
+
+	if u.Scheme != "oss" {
+		return "", "", "", fmt.Errorf("oss: invalid url scheme")
+	}
+
+	// Parse host to extract bucket and region
+	bucket, region, _ = ParseHost(u.Host)
+	if bucket == "" {
+		bucket = u.Host
+	}
+
+	key = strings.TrimPrefix(u.Path, "/")
+	return bucket, region, key, nil
+}
+
+// ParseHost parses the host/endpoint for an OSS storage system.
+// Supports formats like:
+//   - bucket.oss-cn-hangzhou.aliyuncs.com
+//   - bucket.oss-cn-hangzhou-internal.aliyuncs.com
+//   - bucket (just bucket name)
+func ParseHost(host string) (bucket, region, endpoint string) {
+	// Check for internal OSS URL format first (more specific pattern)
+	if a := ossInternalRegex.FindStringSubmatch(host); len(a) > 1 {
+		bucket = a[1]
+		if len(a) > 2 && a[2] != "" {
+			region = a[2]
+		}
+		return bucket, region, ""
+	}
+
+	// Check for standard OSS URL format
+	if a := ossRegex.FindStringSubmatch(host); len(a) > 1 {
+		bucket = a[1]
+		if len(a) > 2 && a[2] != "" {
+			region = a[2]
+		}
+		return bucket, region, ""
+	}
+
+	// For other hosts, assume it's just the bucket name
+	return host, "", ""
+}
+
+var (
+	// oss-cn-hangzhou.aliyuncs.com or bucket.oss-cn-hangzhou.aliyuncs.com
+	ossRegex = regexp.MustCompile(`^(?:([^.]+)\.)?oss-([^.]+)\.aliyuncs\.com$`)
+	// oss-cn-hangzhou-internal.aliyuncs.com or bucket.oss-cn-hangzhou-internal.aliyuncs.com
+	// Uses non-greedy .+? to correctly extract region without -internal suffix
+	ossInternalRegex = regexp.MustCompile(`^(?:([^.]+)\.)?oss-(.+?)-internal\.aliyuncs\.com$`)
+)
+
+func isNotExists(err error) bool {
+	var serviceErr *oss.ServiceError
+	if errors.As(err, &serviceErr) {
+		return serviceErr.Code == "NoSuchKey"
+	}
+	return false
+}
+
+// deleteResultError checks if all requested objects were deleted.
+// OSS SDK doesn't have explicit per-object error reporting like S3, so we verify
+// all requested keys appear in the deleted list.
+func deleteResultError(requested []oss.DeleteObject, out *oss.DeleteMultipleObjectsResult) error {
+	if out == nil {
+		return nil
+	}
+
+	// Build set of deleted keys for quick lookup
+	deleted := make(map[string]struct{}, len(out.DeletedObjects))
+	for _, obj := range out.DeletedObjects {
+		if obj.Key != nil {
+			deleted[*obj.Key] = struct{}{}
+		}
+	}
+
+	// Check that all requested keys were deleted
+	var failed []string
+	for _, obj := range requested {
+		if obj.Key == nil {
+			continue
+		}
+		if _, ok := deleted[*obj.Key]; !ok {
+			failed = append(failed, *obj.Key)
+		}
+	}
+
+	if len(failed) == 0 {
+		return nil
+	}
+
+	// Build error message listing failed keys
+	var b strings.Builder
+	b.WriteString("oss: failed to delete files:")
+	for _, key := range failed {
+		fmt.Fprintf(&b, "\n%s", key)
+	}
+	return errors.New(b.String())
+}
diff --git a/src/litestream-vfs.c b/src/litestream-vfs.c
index f020490..582a651 100644
--- a/src/litestream-vfs.c
+++ b/src/litestream-vfs.c
@@ -1,21 +1,177 @@
 #include "litestream-vfs.h"
 #include "sqlite3.h"
 #include "sqlite3ext.h"
+#include "sqlite3vfs.h"
 #include <stdio.h>
 #include <stdlib.h>
 
 /* sqlite3vfs already called SQLITE_EXTENSION_INIT1 */
 extern const sqlite3_api_routines *sqlite3_api;
 
-// This routine is called when the extension is loaded.
-// Register the new VFS.
+/* Go function declarations */
+extern char* GoLitestreamRegisterConnection(void* db, sqlite3_uint64 file_id);
+extern char* GoLitestreamUnregisterConnection(void* db);
+extern char* GoLitestreamSetTime(void* db, const char* timestamp);
+extern char* GoLitestreamResetTime(void* db);
+extern char* GoLitestreamTime(void* db, char** out);
+extern char* GoLitestreamTxid(void* db, char** out);
+extern char* GoLitestreamLag(void* db, sqlite3_int64* out);
+
+/* Internal function declarations */
+static int litestream_register_connection(sqlite3* db);
+static void litestream_set_time_impl(sqlite3_context* ctx, int argc, sqlite3_value** argv);
+static void litestream_time_impl(sqlite3_context* ctx, int argc, sqlite3_value** argv);
+static void litestream_txid_impl(sqlite3_context* ctx, int argc, sqlite3_value** argv);
+static void litestream_lag_impl(sqlite3_context* ctx, int argc, sqlite3_value** argv);
+static void litestream_function_destroy(void* db);
+static void litestream_auto_extension(sqlite3* db, const char** pzErrMsg, const struct sqlite3_api_routines* pApi);
+
+/* This routine is called when the extension is loaded. */
 int sqlite3_litestreamvfs_init(sqlite3 *db, char **pzErrMsg, const sqlite3_api_routines *pApi) {
   int rc = SQLITE_OK;
   SQLITE_EXTENSION_INIT2(pApi);
 
-  // call into Go to register the VFS
+  /* call into Go to register the VFS */
   LitestreamVFSRegister();
 
+  /* Register SQL functions for new connections. */
+  rc = sqlite3_auto_extension((void (*)(void))litestream_auto_extension);
+
   if( rc==SQLITE_OK ) rc = SQLITE_OK_LOAD_PERMANENTLY;
   return rc;
 }
+
+static void litestream_auto_extension(sqlite3* db, const char** pzErrMsg, const struct sqlite3_api_routines* pApi) {
+  (void)pzErrMsg;
+  (void)pApi;
+
+  if (litestream_register_connection(db) != SQLITE_OK) {
+    return;
+  }
+
+  /* litestream_set_time(timestamp) - for time travel */
+  sqlite3_create_function_v2(db, "litestream_set_time", 1, SQLITE_UTF8 | SQLITE_DIRECTONLY, db, litestream_set_time_impl, 0, 0, litestream_function_destroy);
+
+  /* Read-only functions: litestream_time(), litestream_txid(), litestream_lag() */
+  sqlite3_create_function_v2(db, "litestream_time", 0, SQLITE_UTF8, db, litestream_time_impl, 0, 0, 0);
+  sqlite3_create_function_v2(db, "litestream_txid", 0, SQLITE_UTF8, db, litestream_txid_impl, 0, 0, 0);
+  sqlite3_create_function_v2(db, "litestream_lag", 0, SQLITE_UTF8, db, litestream_lag_impl, 0, 0, 0);
+}
+
+static int litestream_register_connection(sqlite3* db) {
+  sqlite3_file* file = 0;
+  int rc = sqlite3_file_control(db, "main", SQLITE_FCNTL_FILE_POINTER, &file);
+  if (rc != SQLITE_OK || file == 0) {
+    return rc;
+  }
+
+  if (file->pMethods != &s3vfs_io_methods) {
+    /* Not using the litestream VFS. */
+    return SQLITE_DONE;
+  }
+
+  sqlite3_uint64 file_id = ((s3vfsFile*)file)->id;
+  char* err = GoLitestreamRegisterConnection(db, file_id);
+  if (err != 0) {
+    free(err);
+    return SQLITE_ERROR;
+  }
+
+  return SQLITE_OK;
+}
+
+static void litestream_set_time_impl(sqlite3_context* ctx, int argc, sqlite3_value** argv) {
+  if (argc != 1) {
+    sqlite3_result_error(ctx, "expected timestamp argument", -1);
+    return;
+  }
+
+  const unsigned char* ts = sqlite3_value_text(argv[0]);
+  if (!ts) {
+    sqlite3_result_error(ctx, "timestamp required", -1);
+    return;
+  }
+
+  /* Handle special 'LATEST' value */
+  if (sqlite3_stricmp((const char*)ts, "LATEST") == 0) {
+    sqlite3* db = sqlite3_context_db_handle(ctx);
+    char* err = GoLitestreamResetTime(db);
+    if (err != 0) {
+      sqlite3_result_error(ctx, err, -1);
+      free(err);
+      return;
+    }
+    sqlite3_result_null(ctx);
+    return;
+  }
+
+  sqlite3* db = sqlite3_context_db_handle(ctx);
+  char* err = GoLitestreamSetTime(db, (const char*)ts);
+  if (err != 0) {
+    sqlite3_result_error(ctx, err, -1);
+    free(err);
+    return;
+  }
+
+  sqlite3_result_null(ctx);
+}
+
+static void litestream_time_impl(sqlite3_context* ctx, int argc, sqlite3_value** argv) {
+  (void)argc;
+  (void)argv;
+
+  sqlite3* db = sqlite3_context_db_handle(ctx);
+  char* out = 0;
+  char* err = GoLitestreamTime(db, &out);
+  if (err != 0) {
+    sqlite3_result_error(ctx, err, -1);
+    free(err);
+    return;
+  }
+
+  sqlite3_result_text(ctx, out, -1, SQLITE_TRANSIENT);
+  free(out);
+}
+
+static void litestream_txid_impl(sqlite3_context* ctx, int argc, sqlite3_value** argv) {
+  (void)argc;
+  (void)argv;
+
+  sqlite3* db = sqlite3_context_db_handle(ctx);
+  char* out = 0;
+  char* err = GoLitestreamTxid(db, &out);
+  if (err != 0) {
+    sqlite3_result_error(ctx, err, -1);
+    free(err);
+    return;
+  }
+
+  sqlite3_result_text(ctx, out, -1, SQLITE_TRANSIENT);
+  free(out);
+}
+
+static void litestream_lag_impl(sqlite3_context* ctx, int argc, sqlite3_value** argv) {
+  (void)argc;
+  (void)argv;
+
+  sqlite3* db = sqlite3_context_db_handle(ctx);
+  sqlite3_int64 out = 0;
+  char* err = GoLitestreamLag(db, &out);
+  if (err != 0) {
+    sqlite3_result_error(ctx, err, -1);
+    free(err);
+    return;
+  }
+
+  sqlite3_result_int64(ctx, out);
+}
+
+static void litestream_function_destroy(void* db) {
+  if (db == 0) {
+    return;
+  }
+  char* err = GoLitestreamUnregisterConnection(db);
+  if (err != 0) {
+    free(err);
+  }
+}
diff --git a/vfs.go b/vfs.go
index 2984d70..65c78fa 100644
--- a/vfs.go
+++ b/vfs.go
@@ -9,11 +9,14 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
+	_ "unsafe"
 
 	lru "github.com/hashicorp/golang-lru/v2"
+	"github.com/markusmobius/go-dateparser"
 	"github.com/psanford/sqlite3vfs"
 	"github.com/superfly/ltx"
 )
@@ -23,6 +26,16 @@ const (
 	DefaultCacheSize    = 10 * 1024 * 1024 // 10MB
 )
 
+var (
+	//go:linkname sqlite3vfsFileMap github.com/psanford/sqlite3vfs.fileMap
+	sqlite3vfsFileMap map[uint64]sqlite3vfs.File
+
+	//go:linkname sqlite3vfsFileMux github.com/psanford/sqlite3vfs.fileMux
+	sqlite3vfsFileMux sync.Mutex
+
+	vfsConnectionMap sync.Map // map[uintptr]uint64
+)
+
 // VFS implements the SQLite VFS interface for Litestream.
 // It is intended to be used for read replicas that read directly from S3.
 type VFS struct {
@@ -98,13 +111,15 @@ type VFSFile struct {
 	client ReplicaClient
 	name   string
 
-	pos        ltx.Pos  // Last TXID read from level 0 or 1
-	maxTXID1   ltx.TXID // Last TXID read from level 1
-	index      map[uint32]ltx.PageIndexElem
-	pending    map[uint32]ltx.PageIndexElem
-	cache      *lru.Cache[uint32, []byte] // LRU cache for page data
-	targetTime *time.Time                 // Target view time; nil means latest
-	lockType   sqlite3vfs.LockType        // Current lock state
+	pos             ltx.Pos  // Last TXID read from level 0 or 1
+	maxTXID1        ltx.TXID // Last TXID read from level 1
+	index           map[uint32]ltx.PageIndexElem
+	pending         map[uint32]ltx.PageIndexElem
+	cache           *lru.Cache[uint32, []byte] // LRU cache for page data
+	targetTime      *time.Time                 // Target view time; nil means latest
+	latestLTXTime   time.Time                  // Timestamp of most recent LTX file
+	lastPollSuccess time.Time                  // Time of last successful poll
+	lockType        sqlite3vfs.LockType        // Current lock state
 
 	wg     sync.WaitGroup
 	ctx    context.Context
@@ -162,6 +177,20 @@ func (f *VFSFile) LockType() sqlite3vfs.LockType {
 	return f.lockType
 }
 
+// LatestLTXTime returns the timestamp of the most recent LTX file.
+func (f *VFSFile) LatestLTXTime() time.Time {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.latestLTXTime
+}
+
+// LastPollSuccess returns the time of the last successful poll.
+func (f *VFSFile) LastPollSuccess() time.Time {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.lastPollSuccess
+}
+
 func (f *VFSFile) Open() error {
 	f.logger.Info("opening file")
 
@@ -247,6 +276,9 @@ func (f *VFSFile) rebuildIndex(ctx context.Context, infos []*ltx.FileInfo, targe
 	f.pending = make(map[uint32]ltx.PageIndexElem)
 	f.pos = pos
 	f.maxTXID1 = maxTXID1
+	if len(infos) > 0 {
+		f.latestLTXTime = infos[len(infos)-1].CreatedAt
+	}
 	if f.cache != nil {
 		f.cache.Purge()
 	}
@@ -420,6 +452,32 @@ func (f *VFSFile) DeviceCharacteristics() sqlite3vfs.DeviceCharacteristic {
 	return 0
 }
 
+// parseTimeValue parses a timestamp string, trying RFC3339 first, then relative expressions.
+func parseTimeValue(value string) (time.Time, error) {
+	// Try RFC3339Nano first (existing behavior)
+	if t, err := time.Parse(time.RFC3339Nano, value); err == nil {
+		return t, nil
+	}
+
+	// Try RFC3339 (without nanoseconds)
+	if t, err := time.Parse(time.RFC3339, value); err == nil {
+		return t, nil
+	}
+
+	// Fall back to dateparser for relative expressions
+	cfg := &dateparser.Configuration{
+		CurrentTime: time.Now().UTC(),
+	}
+	result, err := dateparser.Parse(cfg, value)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("invalid timestamp (expected RFC3339 or relative time like '5 minutes ago'): %s", value)
+	}
+	if result.Time.IsZero() {
+		return time.Time{}, fmt.Errorf("could not parse time: %s", value)
+	}
+	return result.Time.UTC(), nil
+}
+
 // FileControl handles file control operations, specifically PRAGMA commands for time travel.
 func (f *VFSFile) FileControl(op int, pragmaName string, pragmaValue *string) (*string, error) {
 	const SQLITE_FCNTL_PRAGMA = 14
@@ -433,6 +491,27 @@ func (f *VFSFile) FileControl(op int, pragmaName string, pragmaValue *string) (*
 	f.logger.Info("file control", "pragma", name, "value", pragmaValue)
 
 	switch name {
+	case "litestream_txid":
+		if pragmaValue != nil {
+			return nil, fmt.Errorf("litestream_txid is read-only")
+		}
+		txid := f.Pos().TXID
+		result := txid.String()
+		return &result, nil
+
+	case "litestream_lag":
+		if pragmaValue != nil {
+			return nil, fmt.Errorf("litestream_lag is read-only")
+		}
+		lastPoll := f.LastPollSuccess()
+		if lastPoll.IsZero() {
+			result := "-1" // Never polled successfully
+			return &result, nil
+		}
+		lag := int64(time.Since(lastPoll).Seconds())
+		result := strconv.FormatInt(lag, 10)
+		return &result, nil
+
 	case "litestream_time":
 		if pragmaValue == nil {
 			result := f.currentTimeString()
@@ -446,9 +525,9 @@ func (f *VFSFile) FileControl(op int, pragmaName string, pragmaValue *string) (*
 			return nil, nil
 		}
 
-		t, err := time.Parse(time.RFC3339Nano, *pragmaValue)
+		t, err := parseTimeValue(*pragmaValue)
 		if err != nil {
-			return nil, fmt.Errorf("parse timestamp: %w", err)
+			return nil, err
 		}
 		if err := f.SetTargetTime(context.Background(), t); err != nil {
 			return nil, err
@@ -465,7 +544,10 @@ func (f *VFSFile) currentTimeString() string {
 	if t := f.TargetTime(); t != nil {
 		return t.Format(time.RFC3339Nano)
 	}
-	return "latest"
+	if t := f.LatestLTXTime(); !t.IsZero() {
+		return t.Format(time.RFC3339Nano)
+	}
+	return "latest" // Fallback if no LTX files loaded
 }
 
 func (f *VFSFile) monitorReplicaClient(ctx context.Context) {
@@ -485,6 +567,11 @@ func (f *VFSFile) monitorReplicaClient(ctx context.Context) {
 				if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
 					f.logger.Error("cannot fetch new ltx files", "error", err)
 				}
+			} else {
+				// Track successful poll time
+				f.mu.Lock()
+				f.lastPollSuccess = time.Now()
+				f.mu.Unlock()
 			}
 		}
 	}
@@ -497,14 +584,14 @@ func (f *VFSFile) pollReplicaClient(ctx context.Context) error {
 	index := make(map[uint32]ltx.PageIndexElem)
 	f.logger.Debug("polling replica client", "txid", pos.TXID.String())
 
-	maxTXID0, err := f.pollLevel(ctx, 0, pos.TXID, index)
+	result0, err := f.pollLevel(ctx, 0, pos.TXID, index)
 	if err != nil {
 		return fmt.Errorf("poll L0: %w", err)
 	}
 
-	maxTXID1, err := f.pollLevel(ctx, 1, f.maxTXID1, index)
+	result1, err := f.pollLevel(ctx, 1, f.maxTXID1, index)
 	if err != nil {
-		return fmt.Errorf("poll L0: %w", err)
+		return fmt.Errorf("poll L1: %w", err)
 	}
 
 	// Send updates to a pending list if there are active readers.
@@ -538,10 +625,18 @@ func (f *VFSFile) pollReplicaClient(ctx context.Context) error {
 	}
 
 	// Update to max TXID
-	f.pos.TXID = max(maxTXID0, maxTXID1)
-	f.maxTXID1 = maxTXID1
+	f.pos.TXID = max(result0.maxTXID, result1.maxTXID)
+	f.maxTXID1 = result1.maxTXID
 	f.logger.Debug("txid updated", "txid", f.pos.TXID.String(), "maxTXID1", f.maxTXID1.String())
 
+	// Update latestLTXTime to the most recent timestamp from either level
+	if !result0.createdAt.IsZero() && result0.createdAt.After(f.latestLTXTime) {
+		f.latestLTXTime = result0.createdAt
+	}
+	if !result1.createdAt.IsZero() && result1.createdAt.After(f.latestLTXTime) {
+		f.latestLTXTime = result1.createdAt
+	}
+
 	return nil
 }
 
@@ -561,24 +656,30 @@ func maxLevelTXID(infos []*ltx.FileInfo, level int) ltx.TXID {
 	return maxTXID
 }
 
-func (f *VFSFile) pollLevel(ctx context.Context, level int, prevMaxTXID ltx.TXID, index map[uint32]ltx.PageIndexElem) (ltx.TXID, error) {
+// pollLevelResult contains the results of polling a single level.
+type pollLevelResult struct {
+	maxTXID   ltx.TXID
+	createdAt time.Time // CreatedAt of the most recent LTX file processed
+}
+
+func (f *VFSFile) pollLevel(ctx context.Context, level int, prevMaxTXID ltx.TXID, index map[uint32]ltx.PageIndexElem) (pollLevelResult, error) {
 	// Start reading from the next LTX file after the current position.
 	itr, err := f.client.LTXFiles(ctx, level, prevMaxTXID+1, false)
 	if err != nil {
-		return 0, fmt.Errorf("ltx files: %w", err)
+		return pollLevelResult{}, fmt.Errorf("ltx files: %w", err)
 	}
 
 	// Build an update across all new LTX files.
-	maxTXID := prevMaxTXID
+	result := pollLevelResult{maxTXID: prevMaxTXID}
 	for itr.Next() {
 		info := itr.Item()
 
 		// Ensure we are fetching the next transaction from our current position.
 		f.mu.Lock()
-		isNextTXID := info.MinTXID == maxTXID+1
+		isNextTXID := info.MinTXID == result.maxTXID+1
 		f.mu.Unlock()
 		if !isNextTXID {
-			return maxTXID, fmt.Errorf("non-contiguous ltx file: level=%d, current=%s, next=%s-%s", level, prevMaxTXID, info.MinTXID, info.MaxTXID)
+			return result, fmt.Errorf("non-contiguous ltx file: level=%d, current=%s, next=%s-%s", level, prevMaxTXID, info.MinTXID, info.MaxTXID)
 		}
 
 		f.logger.Debug("new ltx file", "level", info.Level, "min", info.MinTXID, "max", info.MaxTXID)
@@ -586,7 +687,7 @@ func (f *VFSFile) pollLevel(ctx context.Context, level int, prevMaxTXID ltx.TXID
 		// Read page index.
 		idx, err := FetchPageIndex(context.Background(), f.client, info)
 		if err != nil {
-			return maxTXID, fmt.Errorf("fetch page index: %w", err)
+			return result, fmt.Errorf("fetch page index: %w", err)
 		}
 
 		// Update the page index & current position.
@@ -594,8 +695,106 @@ func (f *VFSFile) pollLevel(ctx context.Context, level int, prevMaxTXID ltx.TXID
 			f.logger.Debug("adding new page index", "page", k, "elem", v)
 			index[k] = v
 		}
-		maxTXID = info.MaxTXID
+		result.maxTXID = info.MaxTXID
+		result.createdAt = info.CreatedAt
+	}
+
+	return result, nil
+}
+
+// RegisterVFSConnection maps a SQLite connection handle to its VFS file ID.
+func RegisterVFSConnection(dbPtr uintptr, fileID uint64) error {
+	if _, ok := lookupVFSFile(fileID); !ok {
+		return fmt.Errorf("vfs file not found: id=%d", fileID)
+	}
+	vfsConnectionMap.Store(dbPtr, fileID)
+	return nil
+}
+
+// UnregisterVFSConnection removes a connection mapping.
+func UnregisterVFSConnection(dbPtr uintptr) {
+	vfsConnectionMap.Delete(dbPtr)
+}
+
+// SetVFSConnectionTime rebuilds the VFS index for a connection at a timestamp.
+func SetVFSConnectionTime(dbPtr uintptr, timestamp string) error {
+	file, err := vfsFileForConnection(dbPtr)
+	if err != nil {
+		return err
+	}
+
+	t, err := parseTimeValue(timestamp)
+	if err != nil {
+		return err
+	}
+	return file.SetTargetTime(context.Background(), t)
+}
+
+// ResetVFSConnectionTime rebuilds the VFS index to the latest state.
+func ResetVFSConnectionTime(dbPtr uintptr) error {
+	file, err := vfsFileForConnection(dbPtr)
+	if err != nil {
+		return err
+	}
+	return file.ResetTime(context.Background())
+}
+
+// GetVFSConnectionTime returns the current time for a connection.
+func GetVFSConnectionTime(dbPtr uintptr) (string, error) {
+	file, err := vfsFileForConnection(dbPtr)
+	if err != nil {
+		return "", err
+	}
+	return file.currentTimeString(), nil
+}
+
+// GetVFSConnectionTXID returns the current transaction ID for a connection as a hex string.
+func GetVFSConnectionTXID(dbPtr uintptr) (string, error) {
+	file, err := vfsFileForConnection(dbPtr)
+	if err != nil {
+		return "", err
+	}
+	return file.Pos().TXID.String(), nil
+}
+
+// GetVFSConnectionLag returns seconds since last successful poll for a connection.
+func GetVFSConnectionLag(dbPtr uintptr) (int64, error) {
+	file, err := vfsFileForConnection(dbPtr)
+	if err != nil {
+		return 0, err
+	}
+	lastPoll := file.LastPollSuccess()
+	if lastPoll.IsZero() {
+		return -1, nil
+	}
+	return int64(time.Since(lastPoll).Seconds()), nil
+}
+
+func vfsFileForConnection(dbPtr uintptr) (*VFSFile, error) {
+	v, ok := vfsConnectionMap.Load(dbPtr)
+	if !ok {
+		return nil, fmt.Errorf("connection not registered")
+	}
+	fileID, ok := v.(uint64)
+	if !ok {
+		return nil, fmt.Errorf("invalid connection mapping")
+	}
+	file, ok := lookupVFSFile(fileID)
+	if !ok {
+		return nil, fmt.Errorf("vfs file not found: id=%d", fileID)
+	}
+	return file, nil
+}
+
+func lookupVFSFile(fileID uint64) (*VFSFile, bool) {
+	sqlite3vfsFileMux.Lock()
+	defer sqlite3vfsFileMux.Unlock()
+
+	file, ok := sqlite3vfsFileMap[fileID]
+	if !ok {
+		return nil, false
 	}
 
-	return maxTXID, nil
+	vfsFile, ok := file.(*VFSFile)
+	return vfsFile, ok
 }



Test patch:
diff --git a/cmd/litestream-vfs/main.go b/cmd/litestream-vfs/main.go
index 3664b42..246f3fc 100644
--- a/cmd/litestream-vfs/main.go
+++ b/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
+}
diff --git a/cmd/litestream-vfs/time_travel_test.go b/cmd/litestream-vfs/time_travel_test.go
index 778fe7f..dfc303f 100644
--- a/cmd/litestream-vfs/time_travel_test.go
+++ b/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)
 	}
 }
 
diff --git a/cmd/litestream/main.go b/cmd/litestream/main.go
index 0e8890e..aa749ce 100644
--- a/cmd/litestream/main.go
+++ b/cmd/litestream/main.go
@@ -29,6 +29,7 @@ import (
 	"github.com/benbjohnson/litestream/gs"
 	"github.com/benbjohnson/litestream/internal"
 	"github.com/benbjohnson/litestream/nats"
+	"github.com/benbjohnson/litestream/oss"
 	"github.com/benbjohnson/litestream/s3"
 	"github.com/benbjohnson/litestream/sftp"
 	"github.com/benbjohnson/litestream/webdav"
@@ -1059,6 +1060,10 @@ func NewReplicaFromConfig(c *ReplicaConfig, db *litestream.DB) (_ *litestream.Re
 		if r.Client, err = newNATSReplicaClientFromConfig(c, r); err != nil {
 			return nil, err
 		}
+	case "oss":
+		if r.Client, err = newOSSReplicaClientFromConfig(c, r); err != nil {
+			return nil, err
+		}
 	default:
 		return nil, fmt.Errorf("unknown replica type in config: %q", c.Type)
 	}
@@ -1518,6 +1523,69 @@ func newNATSReplicaClientFromConfig(c *ReplicaConfig, _ *litestream.Replica) (_
 	return client, nil
 }
 
+// newOSSReplicaClientFromConfig returns a new instance of oss.ReplicaClient built from config.
+func newOSSReplicaClientFromConfig(c *ReplicaConfig, _ *litestream.Replica) (_ *oss.ReplicaClient, err error) {
+	// Ensure URL & constituent parts are not both specified.
+	if c.URL != "" && c.Path != "" {
+		return nil, fmt.Errorf("cannot specify url & path for oss replica")
+	} else if c.URL != "" && c.Bucket != "" {
+		return nil, fmt.Errorf("cannot specify url & bucket for oss replica")
+	}
+
+	bucket, configPath := c.Bucket, c.Path
+	region, endpoint := c.Region, c.Endpoint
+
+	// Apply settings from URL, if specified.
+	if c.URL != "" {
+		_, host, upath, err := ParseReplicaURL(c.URL)
+		if err != nil {
+			return nil, err
+		}
+
+		var (
+			ubucket string
+			uregion string
+		)
+
+		ubucket, uregion, _ = oss.ParseHost(host)
+
+		// Only apply URL parts to fields that have not been overridden.
+		if configPath == "" {
+			configPath = upath
+		}
+		if bucket == "" {
+			bucket = ubucket
+		}
+		if region == "" {
+			region = uregion
+		}
+	}
+
+	// Ensure required settings are set.
+	if bucket == "" {
+		return nil, fmt.Errorf("bucket required for oss replica")
+	}
+
+	// Build replica client.
+	client := oss.NewReplicaClient()
+	client.AccessKeyID = c.AccessKeyID
+	client.AccessKeySecret = c.SecretAccessKey
+	client.Bucket = bucket
+	client.Path = configPath
+	client.Region = region
+	client.Endpoint = endpoint
+
+	// Apply upload configuration if specified.
+	if c.PartSize != nil {
+		client.PartSize = int64(*c.PartSize)
+	}
+	if c.Concurrency != nil {
+		client.Concurrency = *c.Concurrency
+	}
+
+	return client, nil
+}
+
 // applyLitestreamEnv copies "LITESTREAM" prefixed environment variables to
 // their AWS counterparts as the "AWS" prefix can be confusing when using a
 // non-AWS S3-compatible service.
diff --git a/cmd/litestream/replicate.go b/cmd/litestream/replicate.go
index e77f3a6..fbe68d5 100644
--- a/cmd/litestream/replicate.go
+++ b/cmd/litestream/replicate.go
@@ -20,6 +20,7 @@ import (
 	"github.com/benbjohnson/litestream/file"
 	"github.com/benbjohnson/litestream/gs"
 	"github.com/benbjohnson/litestream/nats"
+	"github.com/benbjohnson/litestream/oss"
 	"github.com/benbjohnson/litestream/s3"
 	"github.com/benbjohnson/litestream/sftp"
 )
@@ -183,6 +184,8 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 			slogWith.Info("replicating to", "host", client.Host, "user", client.User, "path", client.Path)
 		case *nats.ReplicaClient:
 			slogWith.Info("replicating to", "bucket", client.BucketName, "url", client.URL)
+		case *oss.ReplicaClient:
+			slogWith.Info("replicating to", "bucket", client.Bucket, "path", client.Path, "region", client.Region)
 		default:
 			slogWith.Info("replicating to")
 		}
diff --git a/internal/testingutil/testingutil.go b/internal/testingutil/testingutil.go
index af36998..fb78cbc 100644
--- a/internal/testingutil/testingutil.go
+++ b/internal/testingutil/testingutil.go
@@ -24,6 +24,7 @@ import (
 	"github.com/benbjohnson/litestream/gs"
 	"github.com/benbjohnson/litestream/internal"
 	"github.com/benbjohnson/litestream/nats"
+	"github.com/benbjohnson/litestream/oss"
 	"github.com/benbjohnson/litestream/s3"
 	"github.com/benbjohnson/litestream/sftp"
 	"github.com/benbjohnson/litestream/webdav"
@@ -104,6 +105,16 @@ var (
 	natsPassword = flag.String("nats-password", os.Getenv("LITESTREAM_NATS_PASSWORD"), "")
 )
 
+// Alibaba Cloud OSS settings
+var (
+	ossAccessKeyID     = flag.String("oss-access-key-id", os.Getenv("LITESTREAM_OSS_ACCESS_KEY_ID"), "")
+	ossAccessKeySecret = flag.String("oss-access-key-secret", os.Getenv("LITESTREAM_OSS_ACCESS_KEY_SECRET"), "")
+	ossRegion          = flag.String("oss-region", os.Getenv("LITESTREAM_OSS_REGION"), "")
+	ossBucket          = flag.String("oss-bucket", os.Getenv("LITESTREAM_OSS_BUCKET"), "")
+	ossPath            = flag.String("oss-path", os.Getenv("LITESTREAM_OSS_PATH"), "")
+	ossEndpoint        = flag.String("oss-endpoint", os.Getenv("LITESTREAM_OSS_ENDPOINT"), "")
+)
+
 func Integration() bool {
 	return *integration
 }
@@ -215,6 +226,8 @@ func NewReplicaClient(tb testing.TB, typ string) litestream.ReplicaClient {
 		return NewWebDAVReplicaClient(tb)
 	case nats.ReplicaClientType:
 		return NewNATSReplicaClient(tb)
+	case oss.ReplicaClientType:
+		return NewOSSReplicaClient(tb)
 	case "tigris":
 		return NewTigrisReplicaClient(tb)
 	default:
@@ -347,6 +360,20 @@ func NewNATSReplicaClient(tb testing.TB) *nats.ReplicaClient {
 	return c
 }
 
+// NewOSSReplicaClient returns a new client for integration testing.
+func NewOSSReplicaClient(tb testing.TB) *oss.ReplicaClient {
+	tb.Helper()
+
+	c := oss.NewReplicaClient()
+	c.AccessKeyID = *ossAccessKeyID
+	c.AccessKeySecret = *ossAccessKeySecret
+	c.Region = *ossRegion
+	c.Bucket = *ossBucket
+	c.Path = path.Join(*ossPath, fmt.Sprintf("%016x", rand.Uint64()))
+	c.Endpoint = *ossEndpoint
+	return c
+}
+
 // MustDeleteAll deletes all objects under the client's path.
 func MustDeleteAll(tb testing.TB, c litestream.ReplicaClient) {
 	tb.Helper()
diff --git a/oss/replica_client_test.go b/oss/replica_client_test.go
new file mode 100644
index 0000000..df7518d
--- /dev/null
+++ b/oss/replica_client_test.go
@@ -0,0 +1,372 @@
+package oss
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
+)
+
+func TestReplicaClient_Type(t *testing.T) {
+	c := NewReplicaClient()
+	if got := c.Type(); got != ReplicaClientType {
+		t.Errorf("Type() = %q, want %q", got, ReplicaClientType)
+	}
+	if got := c.Type(); got != "oss" {
+		t.Errorf("Type() = %q, want %q", got, "oss")
+	}
+}
+
+func TestReplicaClient_Init_BucketValidation(t *testing.T) {
+	t.Run("EmptyBucket", func(t *testing.T) {
+		c := NewReplicaClient()
+		c.Bucket = "" // Empty bucket name
+		c.Region = "cn-hangzhou"
+
+		err := c.Init(t.Context())
+		if err == nil {
+			t.Fatal("expected error for empty bucket name")
+		}
+		if got := err.Error(); got != "oss: bucket name is required" {
+			t.Errorf("unexpected error: %v", err)
+		}
+	})
+
+	t.Run("ValidBucketWithRegion", func(t *testing.T) {
+		c := NewReplicaClient()
+		c.Bucket = "test-bucket"
+		c.Region = "cn-hangzhou"
+		c.AccessKeyID = "test-key"
+		c.AccessKeySecret = "test-secret"
+
+		// Init should succeed (client will be created even without real credentials)
+		err := c.Init(t.Context())
+		if err != nil {
+			t.Errorf("Init() should succeed with valid bucket: %v", err)
+		}
+	})
+
+	t.Run("ValidBucketDefaultRegion", func(t *testing.T) {
+		c := NewReplicaClient()
+		c.Bucket = "test-bucket"
+		// Region is empty, should use DefaultRegion
+		c.AccessKeyID = "test-key"
+		c.AccessKeySecret = "test-secret"
+
+		err := c.Init(t.Context())
+		if err != nil {
+			t.Errorf("Init() should succeed with default region: %v", err)
+		}
+	})
+}
+
+func TestReplicaClient_Init_Idempotent(t *testing.T) {
+	c := NewReplicaClient()
+	c.Bucket = "test-bucket"
+	c.AccessKeyID = "test-key"
+	c.AccessKeySecret = "test-secret"
+
+	// First init
+	if err := c.Init(t.Context()); err != nil {
+		t.Fatalf("first Init() failed: %v", err)
+	}
+
+	// Second init should be a no-op
+	if err := c.Init(t.Context()); err != nil {
+		t.Fatalf("second Init() failed: %v", err)
+	}
+}
+
+func TestParseURL(t *testing.T) {
+	tests := []struct {
+		name       string
+		url        string
+		wantBucket string
+		wantRegion string
+		wantKey    string
+		wantErr    bool
+	}{
+		{
+			name:       "SimpleOSSURL",
+			url:        "oss://my-bucket/path/to/file",
+			wantBucket: "my-bucket",
+			wantRegion: "",
+			wantKey:    "path/to/file",
+			wantErr:    false,
+		},
+		{
+			name:       "OSSURLWithRegion",
+			url:        "oss://my-bucket.oss-cn-hangzhou.aliyuncs.com/backup",
+			wantBucket: "my-bucket",
+			wantRegion: "cn-hangzhou",
+			wantKey:    "backup",
+			wantErr:    false,
+		},
+		{
+			name:       "OSSURLNoPath",
+			url:        "oss://my-bucket",
+			wantBucket: "my-bucket",
+			wantRegion: "",
+			wantKey:    "",
+			wantErr:    false,
+		},
+		{
+			name:       "InvalidScheme",
+			url:        "s3://my-bucket/path",
+			wantBucket: "",
+			wantRegion: "",
+			wantKey:    "",
+			wantErr:    true,
+		},
+		{
+			name:       "HTTPScheme",
+			url:        "http://my-bucket/path",
+			wantBucket: "",
+			wantRegion: "",
+			wantKey:    "",
+			wantErr:    true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			bucket, region, key, err := ParseURL(tt.url)
+
+			if (err != nil) != tt.wantErr {
+				t.Errorf("ParseURL() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+
+			if err != nil {
+				return
+			}
+
+			if bucket != tt.wantBucket {
+				t.Errorf("bucket = %q, want %q", bucket, tt.wantBucket)
+			}
+			if region != tt.wantRegion {
+				t.Errorf("region = %q, want %q", region, tt.wantRegion)
+			}
+			if key != tt.wantKey {
+				t.Errorf("key = %q, want %q", key, tt.wantKey)
+			}
+		})
+	}
+}
+
+func TestParseHost(t *testing.T) {
+	tests := []struct {
+		name       string
+		host       string
+		wantBucket string
+		wantRegion string
+	}{
+		{
+			name:       "StandardOSSURL",
+			host:       "my-bucket.oss-cn-hangzhou.aliyuncs.com",
+			wantBucket: "my-bucket",
+			wantRegion: "cn-hangzhou",
+		},
+		{
+			name:       "OSSURLBeijingRegion",
+			host:       "test-bucket.oss-cn-beijing.aliyuncs.com",
+			wantBucket: "test-bucket",
+			wantRegion: "cn-beijing",
+		},
+		{
+			name:       "OSSURLShanghaiRegion",
+			host:       "data-bucket.oss-cn-shanghai.aliyuncs.com",
+			wantBucket: "data-bucket",
+			wantRegion: "cn-shanghai",
+		},
+		{
+			name:       "InternalOSSURL",
+			host:       "my-bucket.oss-cn-hangzhou-internal.aliyuncs.com",
+			wantBucket: "my-bucket",
+			wantRegion: "cn-hangzhou",
+		},
+		{
+			name:       "InternalOSSURLBeijing",
+			host:       "test-bucket.oss-cn-beijing-internal.aliyuncs.com",
+			wantBucket: "test-bucket",
+			wantRegion: "cn-beijing",
+		},
+		{
+			name:       "SimpleBucketName",
+			host:       "my-bucket",
+			wantBucket: "my-bucket",
+			wantRegion: "",
+		},
+		{
+			name:       "BucketWithHyphens",
+			host:       "my-test-bucket-2024.oss-cn-shenzhen.aliyuncs.com",
+			wantBucket: "my-test-bucket-2024",
+			wantRegion: "cn-shenzhen",
+		},
+		{
+			name:       "OSSURLWithNumbers",
+			host:       "bucket123.oss-cn-hangzhou.aliyuncs.com",
+			wantBucket: "bucket123",
+			wantRegion: "cn-hangzhou",
+		},
+		{
+			name:       "OSSURLHongKong",
+			host:       "hk-bucket.oss-cn-hongkong.aliyuncs.com",
+			wantBucket: "hk-bucket",
+			wantRegion: "cn-hongkong",
+		},
+		{
+			name:       "OSSURLSingapore",
+			host:       "sg-bucket.oss-ap-southeast-1.aliyuncs.com",
+			wantBucket: "sg-bucket",
+			wantRegion: "ap-southeast-1",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			bucket, region, _ := ParseHost(tt.host)
+
+			if bucket != tt.wantBucket {
+				t.Errorf("bucket = %q, want %q", bucket, tt.wantBucket)
+			}
+			if region != tt.wantRegion {
+				t.Errorf("region = %q, want %q", region, tt.wantRegion)
+			}
+		})
+	}
+}
+
+func TestIsNotExists(t *testing.T) {
+	t.Run("NilError", func(t *testing.T) {
+		if isNotExists(nil) {
+			t.Error("isNotExists should return false for nil error")
+		}
+	})
+
+	t.Run("RegularError", func(t *testing.T) {
+		regularErr := errors.New("regular error")
+		if isNotExists(regularErr) {
+			t.Error("isNotExists should return false for regular error")
+		}
+	})
+
+	t.Run("WrappedError", func(t *testing.T) {
+		wrappedErr := errors.New("wrapped: something went wrong")
+		if isNotExists(wrappedErr) {
+			t.Error("isNotExists should return false for wrapped non-ServiceError")
+		}
+	})
+}
+
+func TestLtxPath(t *testing.T) {
+	c := NewReplicaClient()
+	c.Path = "backups"
+
+	tests := []struct {
+		level    int
+		filename string
+		want     string
+	}{
+		{0, "00000001-00000001.ltx", "backups/0000/00000001-00000001.ltx"},
+		{1, "00000001-00000010.ltx", "backups/0001/00000001-00000010.ltx"},
+		{15, "00000001-000000ff.ltx", "backups/000f/00000001-000000ff.ltx"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.want, func(t *testing.T) {
+			got := c.ltxPath(tt.level, tt.filename)
+			if got != tt.want {
+				t.Errorf("ltxPath(%d, %q) = %q, want %q", tt.level, tt.filename, got, tt.want)
+			}
+		})
+	}
+}
+
+func TestDeleteResultError(t *testing.T) {
+	ptr := func(s string) *string { return &s }
+
+	t.Run("NilResult", func(t *testing.T) {
+		requested := []oss.DeleteObject{{Key: ptr("key1")}}
+		if err := deleteResultError(requested, nil); err != nil {
+			t.Errorf("expected nil error for nil result, got %v", err)
+		}
+	})
+
+	t.Run("AllDeleted", func(t *testing.T) {
+		requested := []oss.DeleteObject{
+			{Key: ptr("key1")},
+			{Key: ptr("key2")},
+		}
+		result := &oss.DeleteMultipleObjectsResult{
+			DeletedObjects: []oss.DeletedInfo{
+				{Key: ptr("key1")},
+				{Key: ptr("key2")},
+			},
+		}
+		if err := deleteResultError(requested, result); err != nil {
+			t.Errorf("expected nil error when all deleted, got %v", err)
+		}
+	})
+
+	t.Run("SomeNotDeleted", func(t *testing.T) {
+		requested := []oss.DeleteObject{
+			{Key: ptr("key1")},
+			{Key: ptr("key2")},
+			{Key: ptr("key3")},
+		}
+		result := &oss.DeleteMultipleObjectsResult{
+			DeletedObjects: []oss.DeletedInfo{
+				{Key: ptr("key1")},
+				// key2 and key3 not deleted
+			},
+		}
+		err := deleteResultError(requested, result)
+		if err == nil {
+			t.Fatal("expected error when some keys not deleted")
+		}
+		errStr := err.Error()
+		if !contains(errStr, "key2") {
+			t.Errorf("error should mention key2: %s", errStr)
+		}
+		if !contains(errStr, "key3") {
+			t.Errorf("error should mention key3: %s", errStr)
+		}
+	})
+
+	t.Run("EmptyRequested", func(t *testing.T) {
+		requested := []oss.DeleteObject{}
+		result := &oss.DeleteMultipleObjectsResult{}
+		if err := deleteResultError(requested, result); err != nil {
+			t.Errorf("expected nil error for empty requested, got %v", err)
+		}
+	})
+
+	t.Run("NilKeyInRequested", func(t *testing.T) {
+		requested := []oss.DeleteObject{
+			{Key: nil}, // nil key should be skipped
+			{Key: ptr("key1")},
+		}
+		result := &oss.DeleteMultipleObjectsResult{
+			DeletedObjects: []oss.DeletedInfo{
+				{Key: ptr("key1")},
+			},
+		}
+		if err := deleteResultError(requested, result); err != nil {
+			t.Errorf("expected nil error, got %v", err)
+		}
+	})
+}
+
+func contains(s, substr string) bool {
+	return len(s) >= len(substr) && (s == substr || len(s) > 0 && containsHelper(s, substr))
+}
+
+func containsHelper(s, substr string) bool {
+	for i := 0; i <= len(s)-len(substr); i++ {
+		if s[i:i+len(substr)] == substr {
+			return true
+		}
+	}
+	return false
+}