[db-sync] Cap upper limit of sync periodto now - 5s and use DB clock for this

Author: geropl

components/ee/db-sync/src/replication.ts

@@ -70,39 +70,32 @@ export class PeriodicReplicator {
     }
 
     public async synchronize(ignoreStartDate: boolean): Promise<void> {
-        const now = new Date();
-        let previousRun = await this.getLastExportDate();
-        console.info(`Replicating ${this.toString()}: last ran on ${previousRun}`);
-        if (ignoreStartDate) {
-            if (previousRun && previousRun > now) {
-                console.warn(
-                    `Previous run was in the future (${previousRun} > now=${now}). Possible time sync issue between database and db-sync.`,
-                );
-            }
-
-            console.info("Synchronizing complete database (ignoring previous run)");
-            previousRun = undefined;
-        } else if (previousRun && previousRun > now) {
-            throw new Error(
-                `Previous run was in the future (${previousRun} > now=${now}). Possible time sync issue between database and db-sync.`,
-            );
+        let period: SyncPeriod;
+        try {
+            period = await this.getSyncPeriod({ ignoreStartDate, offsetFromNowSeconds: 5 });
+        } catch (err) {
+            console.info("Cannot find a valid sync period; skipping this time.", err);
+            return;
         }
+        console.info(`Replicating: syncing period ${SyncPeriod.toString(period)}`);
 
         const modifications = await this.tableUpdateProvider.getAllStatementsForAllTables(
             this.source,
             this.tableSet,
-            previousRun,
+            period.from,
+            period.to,
         );
         const deletions = modifications.deletions;
         const updates = modifications.updates;
         const total = [...deletions, ...updates];
         console.debug(`Collected ${total.length} statements`);
         try {
-            /* nowait */ this.logStatements(now, total);
+            /* nowait */ this.logStatements(period.to, total);
 
             await Promise.all([this.source, ...this.targets].map((target) => this.update(target, deletions)));
             await Promise.all(this.targets.map((target) => this.update(target, updates)));
-            await this.markLastExportDate(now);
+
+            await this.markLastExportDate(period.to);
         } catch (err) {
             console.error("Error during replication", err);
         }
@@ -175,6 +168,44 @@ export class PeriodicReplicator {
         }
     }
 
+    protected async getSyncPeriod(options: {
+        ignoreStartDate: boolean;
+        offsetFromNowSeconds: number;
+    }): Promise<SyncPeriod> {
+        let previousRun = await this.getLastExportDate();
+        if (options.ignoreStartDate) {
+            console.info("Synchronizing database from beginning of time (ignoring previous run)");
+            previousRun = undefined;
+        }
+
+        const periodEnd = await this.getNextPeriodEnd(options.offsetFromNowSeconds);
+        if (previousRun && periodEnd.getTime() <= previousRun.getTime()) {
+            throw new Error(
+                `Cannot find a period [${previousRun.toISOString()}, now - ${options.offsetFromNowSeconds}s)]`,
+            );
+        }
+
+        return {
+            to: periodEnd,
+            from: previousRun,
+        };
+    }
+
+    protected async getNextPeriodEnd(offsetSeconds: number): Promise<Date> {
+        // End of syncperiod: now - Xs, DB clock
+        const rows = (await query(
+            this.source,
+            // Why CURRENT_TIMESTAMP(3) you ask? Well, because we want to match the precision of getLastExportDate, which is (indirectly) governed by "toISOString" in "markLastExportDate" below.
+            "SELECT SUBTIME(CURRENT_TIMESTAMP(3), SEC_TO_TIME(?)) as upperBound",
+            { values: [offsetSeconds] }, // seconds in the past
+        )) as { upperBound: string | undefined }[];
+        const upperBound = rows[0].upperBound;
+        if (!upperBound) {
+            throw new Error("Unable to retrieve next period end: " + JSON.stringify(rows));
+        }
+        return new Date(upperBound);
+    }
+
     protected async getLastExportDate(): Promise<Date | undefined> {
         try {
             const rows = (await query(
@@ -252,3 +283,15 @@ export class PeriodicReplicator {
         return `${this.source.name} -> [ ${this.targets.map((t) => t.name).join(", ")} ]`;
     }
 }
+
+interface SyncPeriod {
+    to: Date;
+    from: Date | undefined;
+}
+
+namespace SyncPeriod {
+    export function toString(p: SyncPeriod): string {
+        const from = p.from ? p.from.toISOString() : "---";
+        return `${from} -> ${p.to.toISOString()}`;
+    }
+}