Merge 'Partial sync basic' from Nikita Sivukhin

This PR implements basic support for partial sync. Right now the scope
is limited to only `:memory:` IO and later will be properly expanded to
the file based IO later.
The main addition is `PartialDatabaseStorage` which make request to the
remote server for missing local pages on demand.
The main change is that now tursodatabase JS bindings accept optional
"external" IO event loop which in case of sync will drive `ProtocolIo`
internal work associated with remote page fetching tasks.

Closes #3931

Author: penberg

Cargo.lock

@@ -37,13 +37,15 @@ class Database {
   inTransaction: boolean;
 
   private db: NativeDatabase;
+  private ioStep: () => Promise<void>;
   private execLock: AsyncLock;
   private _inTransaction: boolean = false;
   protected connected: boolean = false;
 
-  constructor(db: NativeDatabase) {
+  constructor(db: NativeDatabase, ioStep?: () => Promise<void>) {
     this.db = db;
     this.execLock = new AsyncLock();
+    this.ioStep = ioStep ?? (async () => { });
     Object.defineProperties(this, {
       name: { get: () => this.db.path },
       readonly: { get: () => this.db.readonly },
@@ -74,9 +76,9 @@ class Database {
 
     try {
       if (this.connected) {
-        return new Statement(maybeValue(this.db.prepare(sql)), this.db, this.execLock);
+        return new Statement(maybeValue(this.db.prepare(sql)), this.db, this.execLock, this.ioStep);
       } else {
-        return new Statement(maybePromise(() => this.connect().then(() => this.db.prepare(sql))), this.db, this.execLock)
+        return new Statement(maybePromise(() => this.connect().then(() => this.db.prepare(sql))), this.db, this.execLock, this.ioStep)
       }
     } catch (err) {
       throw convertError(err);
@@ -185,6 +187,7 @@ class Database {
         const stepResult = exec.stepSync();
         if (stepResult === STEP_IO) {
           await this.db.ioLoopAsync();
+          await this.ioStep();
           continue;
         }
         if (stepResult === STEP_DONE) {
@@ -280,11 +283,13 @@ class Statement {
   private stmt: MaybeLazy<NativeStatement>;
   private db: NativeDatabase;
   private execLock: AsyncLock;
+  private ioStep: () => Promise<void>;
 
-  constructor(stmt: MaybeLazy<NativeStatement>, db: NativeDatabase, execLock: AsyncLock) {
+  constructor(stmt: MaybeLazy<NativeStatement>, db: NativeDatabase, execLock: AsyncLock, ioStep: () => Promise<void>) {
     this.stmt = stmt;
     this.db = db;
     this.execLock = execLock;
+    this.ioStep = ioStep;
   }
 
   /**
@@ -352,7 +357,7 @@ class Statement {
       while (true) {
         const stepResult = await stmt.stepSync();
         if (stepResult === STEP_IO) {
-          await this.db.ioLoopAsync();
+          await this.io();
           continue;
         }
         if (stepResult === STEP_DONE) {
@@ -389,7 +394,7 @@ class Statement {
       while (true) {
         const stepResult = await stmt.stepSync();
         if (stepResult === STEP_IO) {
-          await this.db.ioLoopAsync();
+          await this.io();
           continue;
         }
         if (stepResult === STEP_DONE) {
@@ -453,7 +458,7 @@ class Statement {
       while (true) {
         const stepResult = await stmt.stepSync();
         if (stepResult === STEP_IO) {
-          await this.db.ioLoopAsync();
+          await this.io();
           continue;
         }
         if (stepResult === STEP_DONE) {
@@ -471,6 +476,11 @@ class Statement {
     }
   }
 
+  async io() {
+    await this.db.ioLoopAsync();
+    await this.ioStep();
+  }
+
   /**
    * Interrupts the statement.
    */

bindings/javascript/packages/common/promise.ts

@@ -1,7 +1,17 @@
-import { run, memoryIO, SyncEngineGuards } from "./run.js"
-import { DatabaseOpts, ProtocolIo, RunOpts, DatabaseRowMutation, DatabaseRowStatement, DatabaseRowTransformResult, DatabaseStats, DatabaseChangeType, EncryptionOpts } from "./types.js"
+import { run, memoryIO, runner, SyncEngineGuards, Runner } from "./run.js"
+import {
+    DatabaseOpts,
+    ProtocolIo,
+    RunOpts,
+    DatabaseRowMutation,
+    DatabaseRowStatement,
+    DatabaseRowTransformResult,
+    DatabaseStats,
+    DatabaseChangeType,
+    EncryptionOpts,
+} from "./types.js"
 
-export { run, memoryIO, SyncEngineGuards }
+export { run, memoryIO, runner, SyncEngineGuards, Runner }
 export type {
     DatabaseStats,
     DatabaseOpts,

bindings/javascript/sync/packages/common/index.ts

@@ -116,9 +116,28 @@ export function memoryIO(): ProtocolIo {
     }
 };
 
+export interface Runner {
+    wait(): Promise<void>;
+}
 
-export async function run(opts: RunOpts, io: ProtocolIo, engine: any, generator: any): Promise<any> {
+export function runner(opts: RunOpts, io: ProtocolIo, engine: any): Runner {
     let tasks = [];
+    return {
+        async wait() {
+            for (let request = engine.protocolIo(); request != null; request = engine.protocolIo()) {
+                tasks.push(trackPromise(process(opts, io, request)));
+            }
+            const tasksRace = tasks.length == 0 ? Promise.resolve() : Promise.race([timeoutMs(opts.preemptionMs), ...tasks.map(t => t.promise)]);
+            await Promise.all([engine.ioLoopAsync(), tasksRace]);
+
+            tasks = tasks.filter(t => !t.finished);
+
+            engine.protocolIoStep();
+        },
+    }
+}
+
+export async function run(runner: Runner, generator: any): Promise<any> {
     while (true) {
         const { type, ...rest }: GeneratorResponse = await generator.resumeAsync(null);
         if (type == 'Done') {
@@ -131,14 +150,7 @@ export async function run(opts: RunOpts, io: ProtocolIo, engine: any, generator:
             //@ts-ignore
             return rest.changes;
         }
-        for (let request = engine.protocolIo(); request != null; request = engine.protocolIo()) {
-            tasks.push(trackPromise(process(opts, io, request)));
-        }
-
-        const tasksRace = tasks.length == 0 ? Promise.resolve() : Promise.race([timeoutMs(opts.preemptionMs), ...tasks.map(t => t.promise)]);
-        await Promise.all([engine.ioLoopAsync(), tasksRace]);
-
-        tasks = tasks.filter(t => !t.finished);
+        await runner.wait();
     }
 }
 

bindings/javascript/sync/packages/common/run.ts

@@ -107,6 +107,10 @@ export interface DatabaseOpts {
      * optional parameter to enable internal logging for the database
      */
     tracing?: 'error' | 'warn' | 'info' | 'debug' | 'trace',
+    /**
+     * optional parameter to enable partial sync for the database
+     */
+    partialBootstrapStrategy?: { kind: 'prefix', length: number } | { kind: 'query', query: string };
 }
 export interface DatabaseStats {
     /**
@@ -134,6 +138,14 @@ export interface DatabaseStats {
      * (can be used as e-tag, but string must not be interpreted in any way and must be used as opaque value)
      */
     revision: string | null;
+    /**
+     * total amount of sent bytes over the network
+     */
+    networkSentBytes: number;
+    /**
+     * total amount of received bytes over the network
+     */
+    networkReceivedBytes: number;
 }
 
 /* internal types used in the native/browser packages */

bindings/javascript/sync/packages/common/types.ts

@@ -163,7 +163,7 @@ export declare class JsDataCompletion {
 }
 
 export declare class JsProtocolIo {
-  takeRequest(): JsProtocolRequestBytes | null
+
 }
 
 export declare class JsProtocolRequestBytes {
@@ -178,6 +178,7 @@ export declare class SyncEngine {
   /** Runs the I/O loop asynchronously, returning a Promise. */
   ioLoopAsync(): Promise<void>
   protocolIo(): JsProtocolRequestBytes | null
+  protocolIoStep(): void
   push(): GeneratorHolder
   stats(): GeneratorHolder
   wait(): GeneratorHolder
@@ -220,9 +221,13 @@ export type DatabaseRowTransformResultJs =
 export type GeneratorResponse =
   | { type: 'IO' }
   | { type: 'Done' }
-  | { type: 'SyncEngineStats', operations: number, mainWal: number, revertWal: number, lastPullUnixTime?: number, lastPushUnixTime?: number, revision?: string }
+  | { type: 'SyncEngineStats', operations: number, mainWal: number, revertWal: number, lastPullUnixTime?: number, lastPushUnixTime?: number, revision?: string, networkSentBytes: number, networkReceivedBytes: number }
   | { type: 'SyncEngineChanges', changes: SyncEngineChanges }
 
+export type JsPartialBootstrapStrategy =
+  | { type: 'Prefix', length: number }
+  | { type: 'Query', query: string }
+
 export type JsProtocolRequest =
   | { type: 'Http', method: string, path: string, body?: Array<number>, headers: Array<[string, string]> }
   | { type: 'FullRead', path: string }
@@ -240,6 +245,7 @@ export interface SyncEngineOpts {
   protocolVersion?: SyncEngineProtocolVersion
   bootstrapIfEmpty: boolean
   remoteEncryption?: string
+  partialBoostrapStrategy?: JsPartialBootstrapStrategy
 }
 
 export declare const enum SyncEngineProtocolVersion {