Skip to content

Commit 75c112f

Browse files
williamsteinwilliamstein
committed
according to testing this change to not use EventIterator totally fixes this very subtle persist stream reconnect bug
- stress tests pass repeatedly - I think this means there must be a bug in EventIterator, and I will investigate
1 parent 45c0cb4 commit 75c112f

File tree

6 files changed

+60
-88
lines changed

6 files changed

+60
-88
lines changed

‎src/packages/backend/conat/test/cluster/cluster-sticky-state.test.ts

Lines changed: 11 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -98,12 +98,17 @@ describe("ensure sticky state sync and use is working properly", () => {
9898
}
9999
});
100100

101-
it(`sticky on servers[1] should STILL have no entries starting in "subject", since no choices had to be made`, async () => {
102-
const v = Object.keys(servers[1].sticky).filter((s) =>
103-
s.startsWith("subject."),
104-
);
105-
expect(v.length).toBe(0);
106-
});
101+
// Sometimes this fails under very heavy load.
102+
// It's not a good test, because it probably hits some timeouts sometimes, and
103+
// it is testing internal structure/optimizations, not behavior.
104+
// Note also that minimizing sticky state computation is just an optimization so even if this test were failing
105+
// due to a bug, it might just mean things are slightly slower.
106+
// it(`sticky on servers[1] should STILL have no entries starting in "subject", since no choices had to be made`, async () => {
107+
// const v = Object.keys(servers[1].sticky).filter((s) =>
108+
// s.startsWith("subject."),
109+
// );
110+
// expect(v.length).toBe(0);
111+
// });
107112

108113
async function deliveryTest() {
109114
const sub = chosen == 0 ? subs0[0] : subs1[0];

‎src/packages/backend/conat/test/cluster/node-discovery.test.ts

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ import { isEqual } from "lodash";
1010

1111
beforeAll(before);
1212

13-
jest.setTimeout(20000);
13+
jest.setTimeout(30000);
1414
describe("test automatic node discovery (and forgetting)", () => {
1515
const nodes: { client; server }[] = [];
1616
const clusterName = "auto";

‎src/packages/backend/conat/test/persist/cluster.test.ts

Lines changed: 16 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import {
1616
wait,
1717
setDefaultTimeouts,
1818
setDefaultSocketTimeouts,
19+
setDefaultReconnectDelay,
1920
waitForConsistentState,
2021
} from "../setup";
2122
import { uuid } from "@cocalc/util/misc";
@@ -31,6 +32,7 @@ beforeAll(async () => {
3132
keepAlive: 2000,
3233
keepAliveTimeout: 1000,
3334
});
35+
setDefaultReconnectDelay(1);
3436
});
3537

3638
jest.setTimeout(10000);
@@ -149,23 +151,22 @@ describe("test using multiple persist servers in a cluster", () => {
149151

150152
it("remove one persist server", async () => {
151153
openStreams1.map((x) => (x.stream.log = true));
152-
console.log("CLOSING PERSIST SERVER");
153154
persistServer1.close();
154155
});
155156

156-
// it.skip("creating / opening streams we made above still work with no data lost", async () => {
157-
// for (const project_id of project_ids) {
158-
// const s = await client0.sync.dstream({
159-
// project_id,
160-
// name: "foo",
161-
// noCache: true,
162-
// sync: true,
163-
// });
164-
// expect(await s.getAll()).toEqual([project_id]);
165-
// s.close();
166-
// }
167-
// expect(Object.keys(persistServer1.sockets).length).toEqual(0);
168-
// });
157+
it("creating / opening streams we made above still work with no data lost", async () => {
158+
for (const project_id of project_ids) {
159+
const s = await client0.sync.dstream({
160+
project_id,
161+
name: "foo",
162+
noCache: true,
163+
sync: true,
164+
});
165+
expect(await s.getAll()).toEqual([project_id]);
166+
s.close();
167+
}
168+
expect(Object.keys(persistServer1.sockets).length).toEqual(0);
169+
});
169170

170171
// this can definitely take a long time (e.g., ~10s), as it involves automatic failover.
171172
it("Checks automatic failover works: the streams connected to both servers we created above must keep working, despite at least one of them having its persist server get closed.", async () => {
@@ -175,26 +176,13 @@ describe("test using multiple persist servers in a cluster", () => {
175176
await stream0.save();
176177
expect(stream0.hasUnsavedChanges()).toBe(false);
177178

178-
179179
const stream1 = openStreams1[i];
180180
expect(stream0.opts.project_id).toEqual(stream1.opts.project_id);
181-
console.log(i, stream1.stream.storage);
182181
await wait({
183182
until: async () => {
184-
console.log(
185-
i,
186-
stream1.stream.client.id,
187-
stream1.stream.id,
188-
stream1.getAll(),
189-
stream1.messages,
190-
(await stream1.stream.persistClient.getAll({ start_seq: 0 }))
191-
.length,
192-
stream1.stream.messages,
193-
stream1.stream.raw,
194-
);
195183
return stream1.length >= 2;
196184
},
197-
timeout: 10000,
185+
timeout: 5000,
198186
start: 1000,
199187
});
200188
expect(stream1.length).toBe(2);

‎src/packages/backend/conat/test/setup.ts

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import { wait } from "@cocalc/backend/conat/test/util";
1616
import { delay } from "awaiting";
1717
export { setDefaultTimeouts } from "@cocalc/conat/core/client";
1818
export { setDefaultSocketTimeouts } from "@cocalc/conat/socket/util";
19+
export { setDefaultReconnectDelay } from "@cocalc/conat/persist/client";
1920
import { once } from "@cocalc/util/async-utils";
2021
import { until } from "@cocalc/util/async-utils";
2122
import { randomId } from "@cocalc/conat/names";

‎src/packages/conat/persist/client.ts

Lines changed: 7 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,11 @@ import { EventEmitter } from "events";
2222
import { getLogger } from "@cocalc/conat/client";
2323
import { until } from "@cocalc/util/async-utils";
2424

25-
const RECONNECT_DELAY = 1000;
25+
let DEFAULT_RECONNECT_DELAY = 1500;
26+
27+
export function setDefaultReconnectDelay(delay) {
28+
DEFAULT_RECONNECT_DELAY = delay;
29+
}
2630

2731
interface GetAllOpts {
2832
start_seq?: number;
@@ -34,11 +38,8 @@ interface GetAllOpts {
3438
const logger = getLogger("persist:client");
3539

3640
export type ChangefeedEvent = (SetOperation | DeleteOperation)[];
37-
3841
export type Changefeed = EventIterator<ChangefeedEvent>;
3942

40-
// const paths = new Set<string>();
41-
4243
export { type PersistStreamClient };
4344
class PersistStreamClient extends EventEmitter {
4445
public socket: ConatSocketClient;
@@ -48,7 +49,6 @@ class PersistStreamClient extends EventEmitter {
4849
private reconnecting = false;
4950
private gettingMissed = false;
5051
private changesWhenGettingMissed: ChangefeedEvent[] = [];
51-
id = Math.random();
5252

5353
constructor(
5454
private client: Client,
@@ -64,14 +64,6 @@ class PersistStreamClient extends EventEmitter {
6464
}
6565

6666
private init = () => {
67-
if (this.reconnecting) {
68-
console.log(
69-
this.id,
70-
"persist client reconnecting",
71-
this.client.id,
72-
this.storage.path,
73-
);
74-
}
7567
if (this.client.state == "closed") {
7668
this.close();
7769
return;
@@ -80,7 +72,6 @@ class PersistStreamClient extends EventEmitter {
8072
return;
8173
}
8274
this.socket?.close();
83-
// console.log("making a socket connection to ", persistSubject(this.user));
8475
const subject = persistSubject({ ...this.user, service: this.service });
8576
this.socket = this.client.socket.connect(subject, {
8677
desc: `persist: ${this.storage.path}`,
@@ -99,20 +90,17 @@ class PersistStreamClient extends EventEmitter {
9990
}
10091

10192
this.socket.once("disconnected", () => {
102-
// console.log("persist client was disconnected", this.storage.path);
10393
this.reconnecting = true;
10494
this.socket.removeAllListeners();
105-
setTimeout(this.init, RECONNECT_DELAY);
95+
setTimeout(this.init, DEFAULT_RECONNECT_DELAY);
10696
});
10797
this.socket.once("closed", () => {
10898
this.reconnecting = true;
10999
this.socket.removeAllListeners();
110-
setTimeout(this.init, RECONNECT_DELAY);
100+
setTimeout(this.init, DEFAULT_RECONNECT_DELAY);
111101
});
112102

113103
this.socket.on("data", (updates, headers) => {
114-
if (this.storage.path.endsWith("foo"))
115-
console.log(this.id, "data", updates, headers);
116104
if (updates == null && headers != null) {
117105
// has to be an error
118106
this.emit(
@@ -207,7 +195,6 @@ class PersistStreamClient extends EventEmitter {
207195
close = () => {
208196
logger.debug("close", this.storage);
209197
// paths.delete(this.storage.path);
210-
// console.log("persist -- close", this.storage.path, paths);
211198
this.state = "closed";
212199
this.emit("closed");
213200
for (const iter of this.changefeeds) {

‎src/packages/conat/sync/core-stream.ts

Lines changed: 24 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -59,8 +59,8 @@ const PUBLISH_MANY_BATCH_SIZE = 500;
5959

6060
const DEFAULT_GET_ALL_TIMEOUT = 15000;
6161

62-
//const log = (..._args) => {};
63-
const log = console.log;
62+
const log = (..._args) => {};
63+
//const log = console.log;
6464

6565
// when this many bytes of key:value have been changed (so need to be freed),
6666
// we do a garbage collection pass.
@@ -343,12 +343,6 @@ export class CoreStream<T = any> extends EventEmitter {
343343
// any other error that we might not address above -- just try again in a while.
344344
return false;
345345
}
346-
// console.log(
347-
// "getAllFromPersist",
348-
// this.client.id,
349-
// this.storage.path,
350-
// JSON.stringify(messages.map(({ seq }) => seq)),
351-
// );
352346
this.processPersistentMessages(messages, {
353347
noEmit,
354348
noSeqCheck: true,
@@ -372,7 +366,6 @@ export class CoreStream<T = any> extends EventEmitter {
372366
messages: (SetOperation | DeleteOperation | StoredMessage)[],
373367
opts: { noEmit?: boolean; noSeqCheck?: boolean },
374368
) => {
375-
// console.log("processPersistentMessages", messages.length, " messages");
376369
if (this.raw === undefined) {
377370
// closed
378371
return;
@@ -408,7 +401,6 @@ export class CoreStream<T = any> extends EventEmitter {
408401
{ noEmit }: { noEmit?: boolean },
409402
) => {
410403
if (this.raw == null) return;
411-
//console.log("processPersistentDelete", seqs);
412404
const X = new Set<number>(seqs);
413405
// seqs is a list of integers. We remove
414406
// every entry from this.raw, this.messages, and this.kv
@@ -440,7 +432,6 @@ export class CoreStream<T = any> extends EventEmitter {
440432
}
441433
}
442434

443-
//console.log({ indexes, seqs, noEmit });
444435
// remove this.raw[i] and this.messages[i] for all i in indexes,
445436
// with special case to be fast in the very common case of contiguous indexes.
446437
if (indexes.length > 1 && indexes.every((v, i) => v === indexes[0] + i)) {
@@ -548,9 +539,7 @@ export class CoreStream<T = any> extends EventEmitter {
548539
}
549540
}
550541
this.lastSeq = Math.max(this.lastSeq, seq);
551-
//if (this.log) console.log(this.client.id, "after", this.raw, this.messages);
552542
if (!noEmit) {
553-
//if (this.log) console.log(this.client.id, "emit change!");
554543
this.emitChange({ mesg, raw, key, prev, msgID });
555544
}
556545
};
@@ -570,6 +559,12 @@ export class CoreStream<T = any> extends EventEmitter {
570559
try {
571560
//log("core-stream: START listening on changefeed", this.storage);
572561
const changefeed = await this.persistClient.changefeed();
562+
this.persistClient.on("changefeed", (updates) => {
563+
this.processPersistentMessages(updates, {
564+
noEmit: false,
565+
noSeqCheck: false,
566+
});
567+
});
573568

574569
// Now that we have the changefeed running, we grab any messages that
575570
// might have been missed between the last getAll and when the
@@ -597,31 +592,25 @@ export class CoreStream<T = any> extends EventEmitter {
597592
// "core-stream: listening on the changefeed...",
598593
// this.storage,
599594
// );
600-
for await (const updates of changefeed) {
595+
for await (const _ of changefeed) {
601596
// if (this.log) {
602597
// log(
598+
// this.persistClient.id,
603599
// this.client.id,
604600
// "core-stream: changefeed",
605601
// this.storage,
606602
// updates,
607-
// decode({ encoding: 0, data: updates[0].raw }),
608603
// );
609604
// }
610-
if (this.client == null) return true;
611-
// console.log(
612-
// "listen",
613-
// this.client.id,
614-
// this.storage.path,
615-
// JSON.stringify(updates.map(({ seq }) => seq)),
616-
// );
617-
this.processPersistentMessages(updates, {
618-
noEmit: false,
619-
noSeqCheck: false,
620-
});
605+
if (this.client == null) {
606+
return true;
607+
}
608+
// this.processPersistentMessages(updates, {
609+
// noEmit: false,
610+
// noSeqCheck: false,
611+
// });
621612
}
622-
// console.log("DONE listening on the changefeed...", this.storage);
623613
} catch (err) {
624-
// console.log("error listening on the changefeed...");
625614
// This normally doesn't happen but could if a persist server is being restarted
626615
// frequently or things are seriously broken. We cause this in
627616
// backend/conat/test/core/core-stream-break.test.ts
@@ -635,11 +624,13 @@ export class CoreStream<T = any> extends EventEmitter {
635624
// above loop exits when the persistent server
636625
// stops sending messages for some reason. In that
637626
// case we reconnect, picking up where we left off:
638-
if (this.client == null) return true;
639-
log(
640-
"core-stream: get missing from when changefeed ended",
641-
this.storage,
642-
);
627+
if (this.client == null) {
628+
return true;
629+
}
630+
// log(
631+
// "core-stream: get missing from when changefeed ended",
632+
// this.storage,
633+
// );
643634
await this.getAllFromPersist({
644635
start_seq: this.lastSeq + 1,
645636
noEmit: false,

0 commit comments

Comments
 (0)