Skip to content

Commit 49c9682

Browse files
committed
Add Replica.run() continuous sync pattern and document it
Add Replica.run({ intervalMs, signal }) which syncs immediately and then repeats on an interval until the AbortSignal is aborted; transient sync errors are logged and the loop continues. Add continuous-sync tests (picks up new writes, survives a transient write failure via a flaky client). Document the pattern in the README and architecture notes and fix a duplicate section heading.
1 parent c012d7e commit 49c9682

4 files changed

Lines changed: 202 additions & 4 deletions

File tree

‎README.md‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -117,6 +117,38 @@ const client = newReplicaClientFromURL('file:///data/replica');
117117
const replica = new Replica('/data/app.db', client);
118118
```
119119

120+
### Continuous sync
121+
122+
For ongoing replication, `replica.run()` syncs immediately and then repeats on an
123+
interval until the supplied `AbortSignal` is aborted. Sync errors are logged and the
124+
loop keeps going, so a transient failure doesn't stop replication:
125+
126+
```ts
127+
import { Replica, FileReplicaClient } from 'sqlite-js-stream';
128+
129+
const replica = new Replica('/data/app.db', new FileReplicaClient('/data/replica'));
130+
const ac = new AbortController();
131+
132+
// Run continuously (sync now, then every 1s); resolves when aborted.
133+
const runP = replica.run({ intervalMs: 1000, signal: ac.signal });
134+
135+
// ... later, to stop:
136+
ac.abort();
137+
await runP;
138+
```
139+
140+
If you want to keep replicating in the background without awaiting, call `run()` and
141+
`catch` its promise (it resolves on abort):
142+
143+
```ts
144+
replica.run({ intervalMs: 1000, signal: ac.signal }).catch((err) => {
145+
console.error('continuous sync stopped', err);
146+
});
147+
```
148+
149+
`Store` also exposes `syncAll()`/`snapshotAll()` for one-shot operations; a continuous
150+
loop can be driven by calling `replica.run()` per database, or via your own scheduler.
151+
120152
---
121153

122154
## How it works

‎docs/ARCHITECTURE.md‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -66,11 +66,13 @@ prevents the WAL from being checkpointed past our position.
6666
Writes a **level-9** full snapshot LTX file (`MinTXID = 1`, `MaxTXID = current position`) containing
6767
every committed page. Combined with incremental level-0 files this gives point-in-time restore.
6868

69-
### Replica.sync()
69+
### Replica.run()
7070

71-
Uploads every local level-0 LTX file from `replicaPos + 1` to `dbPos` to the client, advancing the
72-
replicated position as it goes. The replica position is the max `MaxTXID` among level-0 files on the
73-
client.
71+
`Replica.sync()` writes a local level-0 LTX file and then uploads every pending L0 file from
72+
`replicaPos + 1` to `dbPos`, advancing the replicated position (which is the max `MaxTXID` among
73+
level-0 files on the client). `Replica.run({ intervalMs, signal })` loops this continuously until
74+
the `AbortSignal` is aborted: it syncs immediately, waits `intervalMs`, then repeats. A transient
75+
sync error is logged and the loop keeps going.
7476

7577
### Replica.restore()
7678

‎src/replica.ts‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818

1919
import { access, mkdir, rename, writeFile } from 'node:fs/promises';
2020
import nodePath from 'node:path';
21+
import { setTimeout as sleep } from 'node:timers/promises';
2122

2223
import {
2324
BufferSink,
@@ -44,6 +45,13 @@ export interface SyncStatus {
4445
inSync: boolean;
4546
}
4647

48+
export interface ReplicaRunOptions {
49+
/** Milliseconds between syncs. Defaults to 1000. */
50+
intervalMs?: number;
51+
/** Abort to stop continuous syncing. */
52+
signal?: AbortSignal;
53+
}
54+
4755
function chunks(data: Uint8Array, size = 256 * 1024): AsyncIterable<Uint8Array> {
4856
return (async function* () {
4957
for (let i = 0; i < data.length; i += size) {
@@ -120,6 +128,29 @@ export class Replica {
120128
await this.sync();
121129
}
122130

131+
/**
132+
* Run continuously syncs the replica until `signal` is aborted. It syncs
133+
* immediately, then waits `intervalMs` and repeats. Errors during a sync are
134+
* logged and the loop keeps going, so a transient failure does not stop
135+
* replication. Resolves once the signal is aborted.
136+
*/
137+
async run(opt: ReplicaRunOptions = {}): Promise<void> {
138+
const { intervalMs = 1000, signal } = opt;
139+
while (!signal?.aborted) {
140+
try {
141+
await this.sync();
142+
} catch (err) {
143+
this.logger.error('replica sync error', err);
144+
}
145+
if (signal?.aborted) break;
146+
try {
147+
await sleep(intervalMs, undefined, { signal });
148+
} catch {
149+
break; // aborted during the sleep
150+
}
151+
}
152+
}
153+
123154
/** Snapshot writes and uploads a level-9 full-database LTX file. */
124155
async snapshot(): Promise<void> {
125156
const info = await (await this.localDb()).snapshot();
Lines changed: 133 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,133 @@
1+
import { test } from 'node:test';
2+
import assert from 'node:assert/strict';
3+
import { DatabaseSync } from 'node:sqlite';
4+
import type { Readable } from 'node:stream';
5+
import { join } from 'node:path';
6+
7+
import {
8+
Replica,
9+
FileReplicaClient,
10+
type ReplicaClient,
11+
type FileInfo,
12+
type LTXFilesOptions,
13+
type OpenLTXFileOptions,
14+
type WriteLTXFileOptions,
15+
} from '../../src/index.ts';
16+
import type { Logger } from '../../src/util/logger.ts';
17+
import { createWALDB, makeTempDir } from '../helpers/make-db.ts';
18+
19+
async function waitUntil(fn: () => Promise<boolean>, timeoutMs = 8000): Promise<void> {
20+
const start = Date.now();
21+
while (Date.now() - start < timeoutMs) {
22+
if (await fn()) return;
23+
await new Promise((r) => setTimeout(r, 5));
24+
}
25+
throw new Error('timed out waiting for condition');
26+
}
27+
28+
/** A ReplicaClient that fails the first write, then delegates. */
29+
class FlakyOnceClient implements ReplicaClient {
30+
private inner: FileReplicaClient;
31+
private failed = false;
32+
constructor(inner: FileReplicaClient) {
33+
this.inner = inner;
34+
}
35+
type(): string {
36+
return this.inner.type();
37+
}
38+
init(): Promise<void> {
39+
return this.inner.init();
40+
}
41+
ltxFiles(opts: LTXFilesOptions): AsyncIterable<FileInfo> {
42+
return this.inner.ltxFiles(opts);
43+
}
44+
openLtxFile(opts: OpenLTXFileOptions): Promise<Readable> {
45+
return this.inner.openLtxFile(opts);
46+
}
47+
async writeLtxFile(
48+
opts: WriteLTXFileOptions,
49+
body: AsyncIterable<Uint8Array | string>,
50+
): Promise<FileInfo> {
51+
if (!this.failed) {
52+
this.failed = true;
53+
throw new Error('transient write failure');
54+
}
55+
return this.inner.writeLtxFile(opts, body);
56+
}
57+
deleteLtxFiles(infos: FileInfo[]): Promise<void> {
58+
return this.inner.deleteLtxFiles(infos);
59+
}
60+
deleteAll(): Promise<void> {
61+
return this.inner.deleteAll();
62+
}
63+
setLogger(logger: Logger): void {
64+
this.inner.setLogger(logger);
65+
}
66+
}
67+
68+
test('run() continuously syncs new writes until aborted', async () => {
69+
const t = createWALDB(['a', 'b', 'c']);
70+
const client = new FileReplicaClient(makeTempDir('repl-'));
71+
const replica = new Replica(t.path, client);
72+
const ac = new AbortController();
73+
74+
const runP = replica.run({ intervalMs: 25, signal: ac.signal });
75+
try {
76+
// Let the first sync run, then add more data.
77+
await new Promise((r) => setTimeout(r, 30));
78+
t.db.exec("INSERT INTO t (name) VALUES ('d'), ('e'), ('f')");
79+
80+
// The continuous loop should pick up the new rows on a later iteration.
81+
await waitUntil(async () => (await replica.syncStatus()).localTXID >= 2n);
82+
83+
ac.abort();
84+
await runP;
85+
86+
const out = join(makeTempDir('restore-'), 'db.db');
87+
await replica.restore({ outputPath: out });
88+
const r = new DatabaseSync(out);
89+
const count = r.prepare('SELECT count(*) AS n FROM t').get() as { n: number };
90+
assert.equal(count.n, 6);
91+
const ic = r.prepare('PRAGMA integrity_check').get() as { integrity_check: string };
92+
assert.equal(ic.integrity_check, 'ok');
93+
r.close();
94+
} finally {
95+
ac.abort();
96+
replica.close();
97+
t.close();
98+
}
99+
});
100+
101+
test('run() keeps replicating after a transient write failure', async () => {
102+
const t = createWALDB(['a', 'b']);
103+
const client = new FlakyOnceClient(new FileReplicaClient(makeTempDir('repl-')));
104+
const replica = new Replica(t.path, client);
105+
const ac = new AbortController();
106+
107+
const runP = replica.run({ intervalMs: 25, signal: ac.signal });
108+
try {
109+
// Add data; the first upload will fail, but the loop retries and recovers.
110+
await new Promise((r) => setTimeout(r, 30));
111+
t.db.exec("INSERT INTO t (name) VALUES ('c'), ('d')");
112+
113+
// localTXID advances on the local write; remoteTXID >= 2 means the retry
114+
// eventually uploaded everything despite the transient failure.
115+
await waitUntil(async () => (await replica.syncStatus()).remoteTXID >= 2n);
116+
117+
ac.abort();
118+
await runP;
119+
120+
const out = join(makeTempDir('restore-'), 'db.db');
121+
await replica.restore({ outputPath: out });
122+
const r = new DatabaseSync(out);
123+
const count = r.prepare('SELECT count(*) AS n FROM t').get() as { n: number };
124+
assert.equal(count.n, 4);
125+
const ic = r.prepare('PRAGMA integrity_check').get() as { integrity_check: string };
126+
assert.equal(ic.integrity_check, 'ok');
127+
r.close();
128+
} finally {
129+
ac.abort();
130+
replica.close();
131+
t.close();
132+
}
133+
});

0 commit comments

Comments
 (0)