Skip to content

Commit 544e784

Browse files
geroplroboquat
geropl
authored and
roboquat
committed
[db-sync] Fix it
1 parent 23419be commit 544e784

File tree

2 files changed

+96
-58
lines changed

2 files changed

+96
-58
lines changed

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

+95-57
Original file line numberDiff line numberDiff line change
@@ -9,11 +9,16 @@ import { TableUpdateProvider } from "./export";
99
import * as ProgressBar from "progress";
1010
import { query, NamedConnection } from "./database";
1111
import { injectable, inject } from "inversify";
12-
import * as path from 'path';
13-
import * as fs from 'fs';
14-
import { Semaphore } from '@gitpod/gitpod-protocol/lib/util/semaphore'
15-
16-
export type PeriodicReplicatorProvider = (source: Connection, targets: Connection[], syncInterval: number, tableSet: string) => PeriodicReplicator
12+
import * as path from "path";
13+
import * as fs from "fs";
14+
import { Semaphore } from "@gitpod/gitpod-protocol/lib/util/semaphore";
15+
16+
export type PeriodicReplicatorProvider = (
17+
source: Connection,
18+
targets: Connection[],
19+
syncInterval: number,
20+
tableSet: string,
21+
) => PeriodicReplicator;
1722
export const PeriodicReplicatorProvider = Symbol("PeriodicReplicatorProvider");
1823

1924
@injectable()
@@ -31,7 +36,12 @@ export class PeriodicReplicator {
3136

3237
// This is a weird setup and I'd rather have those fields set in the constructor.
3338
// I have not found a way how to do that using inversify.
34-
public setup(source: NamedConnection, targets: NamedConnection[], syncInterval: number, tableSet: string | undefined) {
39+
public setup(
40+
source: NamedConnection,
41+
targets: NamedConnection[],
42+
syncInterval: number,
43+
tableSet: string | undefined,
44+
) {
3545
this.source = source;
3646
this.targets = targets;
3747
this.syncInterval = syncInterval;
@@ -63,29 +73,37 @@ export class PeriodicReplicator {
6373
const now = new Date();
6474
let previousRun = await this.getLastExportDate();
6575
console.info(`Replicating ${this.toString()}: last ran on ${previousRun}`);
66-
if(ignoreStartDate) {
67-
if(previousRun && previousRun > now) {
68-
console.warn(`Previous run was in the future (${previousRun} > now=${now}). Possible time sync issue between database and db-sync.`);
76+
if (ignoreStartDate) {
77+
if (previousRun && previousRun > now) {
78+
console.warn(
79+
`Previous run was in the future (${previousRun} > now=${now}). Possible time sync issue between database and db-sync.`,
80+
);
6981
}
7082

7183
console.info("Synchronizing complete database (ignoring previous run)");
7284
previousRun = undefined;
73-
} else if(previousRun && previousRun > now) {
74-
throw new Error(`Previous run was in the future (${previousRun} > now=${now}). Possible time sync issue between database and db-sync.`);
85+
} else if (previousRun && previousRun > now) {
86+
throw new Error(
87+
`Previous run was in the future (${previousRun} > now=${now}). Possible time sync issue between database and db-sync.`,
88+
);
7589
}
7690

77-
const modifications = await this.tableUpdateProvider.getAllStatementsForAllTables(this.source, this.tableSet, previousRun);
91+
const modifications = await this.tableUpdateProvider.getAllStatementsForAllTables(
92+
this.source,
93+
this.tableSet,
94+
previousRun,
95+
);
7896
const deletions = modifications.deletions;
7997
const updates = modifications.updates;
8098
const total = [...deletions, ...updates];
81-
console.debug(`Collected ${total.length} statements`)
99+
console.debug(`Collected ${total.length} statements`);
82100
try {
83101
/* nowait */ this.logStatements(now, total);
84102

85-
await Promise.all([ this.source, ...this.targets ].map(target => this.update(target, deletions)));
86-
await Promise.all(this.targets.map(target => this.update(target, updates)));
103+
await Promise.all([this.source, ...this.targets].map((target) => this.update(target, deletions)));
104+
await Promise.all(this.targets.map((target) => this.update(target, updates)));
87105
await this.markLastExportDate(now);
88-
} catch(err) {
106+
} catch (err) {
89107
console.error("Error during replication", err);
90108
}
91109
}
@@ -94,7 +112,7 @@ export class PeriodicReplicator {
94112
if (!this.logdir) return;
95113

96114
const logdir = this.logdir;
97-
const dest = path.join(logdir, `${this.tableSet || 'default'}-${now.getTime()}.sql`);
115+
const dest = path.join(logdir, `${this.tableSet || "default"}-${now.getTime()}.sql`);
98116
try {
99117
if (!(await new Promise<boolean>((resolve, reject) => fs.exists(logdir, resolve)))) {
100118
await new Promise<void>((resolve, reject) => {
@@ -105,20 +123,20 @@ export class PeriodicReplicator {
105123
resolve();
106124
}
107125
});
108-
})
126+
});
109127
}
110128

111-
const logfile = fs.createWriteStream(dest, { flags: 'w' });
129+
const logfile = fs.createWriteStream(dest, { flags: "w" });
112130
const semaphore = new Semaphore(1);
113-
logfile.on('drain', () => semaphore.release());
131+
logfile.on("drain", () => semaphore.release());
114132
for (const row of updates) {
115133
const written = logfile.write(row + "\n");
116134
if (!written) {
117135
await semaphore.acquire();
118136
}
119137
}
120138
console.debug(`Log file ${dest} written`);
121-
} catch(err) {
139+
} catch (err) {
122140
console.warn("Error while writing log file to " + dest, err);
123141
}
124142

@@ -127,23 +145,27 @@ export class PeriodicReplicator {
127145

128146
protected async deleteOldLogs(logdir: string) {
129147
try {
130-
const files = await new Promise<string[]>((resolve, reject) => fs.readdir(this.logdir!, (err: any, files: string[]) => {
131-
if (err) {
132-
reject(err);
133-
} else {
134-
resolve(files)
135-
}
136-
}));
148+
const files = await new Promise<string[]>((resolve, reject) =>
149+
fs.readdir(this.logdir!, (err: any, files: string[]) => {
150+
if (err) {
151+
reject(err);
152+
} else {
153+
resolve(files);
154+
}
155+
}),
156+
);
137157
for (const file of files) {
138158
// We don't care about errors during deletion: it's racy anyway (see "nowait" above), and will succeed next time
139159
const filePath = path.join(logdir, file);
140-
const ctime = await new Promise<number>((resolve, reject) => fs.stat(filePath, (err, stats) => {
141-
if (!err) {
142-
resolve(stats.ctimeMs);
143-
}
144-
}));
160+
const ctime = await new Promise<number>((resolve, reject) =>
161+
fs.stat(filePath, (err, stats) => {
162+
if (!err) {
163+
resolve(stats.ctimeMs);
164+
}
165+
}),
166+
);
145167
const now = Date.now();
146-
const endTime = ctime + (2 * 24 * 60 * 60);
168+
const endTime = ctime + 2 * 24 * 60 * 60;
147169
if (now > endTime) {
148170
fs.unlink(filePath, (_) => {});
149171
}
@@ -155,13 +177,16 @@ export class PeriodicReplicator {
155177

156178
protected async getLastExportDate(): Promise<Date | undefined> {
157179
try {
158-
const rows = await query(this.source, "SELECT value FROM gitpod_replication WHERE item = 'lastExport'") as any[];
159-
if(rows.length > 0) {
160-
return new Date(rows[0]['value'] as string);
180+
const rows = (await query(
181+
this.source,
182+
"SELECT value FROM gitpod_replication WHERE item = 'lastExport'",
183+
)) as any[];
184+
if (rows.length > 0) {
185+
return new Date(rows[0]["value"] as string);
161186
}
162187
return undefined;
163-
} catch(err) {
164-
if(err.toString().indexOf("ER_NO_SUCH_TABLE") > -1) {
188+
} catch (err) {
189+
if (err.toString().indexOf("ER_NO_SUCH_TABLE") > -1) {
165190
return undefined;
166191
} else {
167192
throw err;
@@ -170,32 +195,39 @@ export class PeriodicReplicator {
170195
}
171196

172197
protected async update(target: Connection, updates: string[], batchSize = 100) {
173-
const updateSize = updates.join().length;
174-
const inTransaction = updateSize < (8 * 1024 * 1024) && this.useTransactions;
175-
if(inTransaction) {
176-
await query(target, "START TRANSACTION;");
177-
} else {
178-
console.warn("Update is too big (> 8mb) or transactions are disabled, not running in a transaction! Inconsistency is possible.");
179-
}
180-
181-
const bar = this.showProgressBar && (updates.length > batchSize) ? new ProgressBar('inserting/updating [:bar] :rate/bps :percent :etas', updates.length / batchSize) : { tick: () => {}, terminate: () => {} };
198+
// const updateSize = updates.join().length;
199+
// const inTransaction = updateSize < (8 * 1024 * 1024) && this.useTransactions;
200+
// if(inTransaction) {
201+
// await query(target, "START TRANSACTION;");
202+
// } else {
203+
// console.warn("Update is too big (> 8mb) or transactions are disabled, not running in a transaction! Inconsistency is possible.");
204+
// }
205+
const inTransaction = false;
206+
207+
const bar =
208+
this.showProgressBar && updates.length > batchSize
209+
? new ProgressBar("inserting/updating [:bar] :rate/bps :percent :etas", updates.length / batchSize)
210+
: { tick: () => {}, terminate: () => {} };
182211
try {
183-
for(var i = 0; i < updates.length; i += batchSize) {
212+
for (var i = 0; i < updates.length; i += batchSize) {
184213
const imax = Math.min(i + batchSize, updates.length);
185214
const thisUpdate = updates.slice(i, imax).join("");
186215
await query(target, thisUpdate);
187216
bar.tick();
188217
}
189-
if(inTransaction) {
218+
if (inTransaction) {
190219
console.debug("Modifications were OK. Committing transaction.");
191220
await query(target, "COMMIT;");
192221
}
193-
} catch(err) {
194-
if(inTransaction) {
222+
} catch (err) {
223+
if (inTransaction) {
195224
console.error("Caught an error during modification. Rolling back transaction.", err);
196225
await query(target, "ROLLBACK;");
197226
} else {
198-
console.error("Caught an error during modification. NOT RUNNING IN A TRANSACTION. Data may be inconsistent.", err);
227+
console.error(
228+
"Caught an error during modification. NOT RUNNING IN A TRANSACTION. Data may be inconsistent.",
229+
err,
230+
);
199231
}
200232
throw err;
201233
} finally {
@@ -204,12 +236,18 @@ export class PeriodicReplicator {
204236
}
205237

206238
async markLastExportDate(date: Date) {
207-
await query(this.source, "CREATE TABLE IF NOT EXISTS gitpod_replication (item VARCHAR(36), value VARCHAR(255), PRIMARY KEY (item))");
208-
await query(this.source, "INSERT INTO gitpod_replication VALUES ('lastExport', ?) ON DUPLICATE KEY UPDATE value=?", { values: [ date.toISOString(), date.toISOString() ] });
239+
await query(
240+
this.source,
241+
"CREATE TABLE IF NOT EXISTS gitpod_replication (item VARCHAR(36), value VARCHAR(255), PRIMARY KEY (item))",
242+
);
243+
await query(
244+
this.source,
245+
"INSERT INTO gitpod_replication VALUES ('lastExport', ?) ON DUPLICATE KEY UPDATE value=?",
246+
{ values: [date.toISOString(), date.toISOString()] },
247+
);
209248
}
210249

211250
public toString(): string {
212-
return `${this.source.name} -> [ ${this.targets.map(t => t.name).join(', ')} ]`;
251+
return `${this.source.name} -> [ ${this.targets.map((t) => t.name).join(", ")} ]`;
213252
}
214-
215253
}

‎components/gitpod-db/src/tables.ts

+1-1
Original file line numberDiff line numberDiff line change
@@ -282,7 +282,7 @@ export class GitpodTableDescriptionProvider implements TableDescriptionProvider
282282
},
283283
{
284284
name: "d_b_cost_center",
285-
primaryKeys: ["id"],
285+
primaryKeys: ["id", "creationTime"],
286286
deletionColumn: "deleted",
287287
timeColumn: "_lastModified",
288288
},

0 commit comments

Comments
 (0)