1
/*
2
Unit tests of multiple persist servers at once:
3

4
- making numerous distinct clients and seeing that they are distributed across persist servers
5
- stopping a persist server and seeing that failover happens without data loss
6

7
pnpm test `pwd`/multiple-servers.test.ts 
8

9
*/
10

11
import {
12
  before,
13
  after,
14
  connect,
15
  persistServer as defaultPersistServer,
16
  once,
17
} from "@cocalc/backend/conat/test/setup";
18
import { stream } from "@cocalc/conat/persist/client";
19
import { delay } from "awaiting";
20
import { server as createPersistServer } from "@cocalc/backend/conat/persist";
21
import { messageData } from "@cocalc/conat/core/client";
22
import { uuid } from "@cocalc/util/misc";
23

24
beforeAll(before);
25

26
describe("multiple clients using multiple persist servers", () => {
27
  const persistServers: any[] = [];
28
  let numServers = 4;
29
  it(`ceate ${numServers} persist servers`, async () => {
30
    persistServers.push(defaultPersistServer);
31
    for (let i = 0; i < numServers - 1; i++) {
32
      const client = connect();
33
      const persistServer = createPersistServer({ client });
34
      await once(persistServer, "ready");
35
      persistServers.push(persistServer);
36
    }
37
  });
38

39
  let persistClients: any[] = [];
40
  
41
  // NOTE: count must be below about 40 to avoid hitting the default
42
  // per-user connection limit of 100.
43
  let count = 30;
44
  const projects: string[] = [];
45
  it(`creates ${count} persist clients`, async () => {
46
    const ids = new Set<string>([]);
47
    for (let i = 0; i < count; i++) {
48
      const client = connect();
49
      const project_id = uuid();
50
      projects.push(project_id);
51
      const persistClient = stream({
52
        client,
53
        user: { project_id },
54
        storage: { path: `projects/${project_id}/foo-${i}` },
55
      });
56
      ids.add(await persistClient.serverId());
57
      persistClients.push(persistClient);
58
      const { seq } = await persistClient.set({
59
        messageData: messageData(i, { headers: { [i]: i } }),
60
      });
61
      expect(seq).toBe(1);
62
    }
63
    // given that we're randomly distributing so many clients (what really matters
64
    // is the user field above)
65
    // it's highly likely we hit all servers.
66
    expect(ids.size).toBe(persistServers.length);
67
  });
68

69
  it(`add ${numServers} more persist servers`, async () => {
70
    for (let i = 0; i < numServers - 1; i++) {
71
      const client = connect();
72
      const persistServer = createPersistServer({ client });
73
      await once(persistServer, "ready");
74
      persistServers.push(persistServer);
75
    }
76
  });
77

78
  it("read data we wrote above (so having new servers doesn't mess with existing connections)", async () => {
79
    for (let i = 0; i < count; i++) {
80
      const mesg = await persistClients[i].get({ seq: 1 });
81
      expect(mesg.data).toBe(i);
82
      expect(mesg.headers[`${i}`]).toBe(i);
83
    }
84
  });
85

86
  it("new clients use exactly the same servers, since the assignment was already made above", async () => {
87
    const ids = new Set<string>([]);
88
    for (let i = 0; i < count; i++) {
89
      const client = connect();
90
      const project_id = projects[i];
91
      const persistClient = stream({
92
        client,
93
        user: { project_id },
94
        storage: { path: `projects/${project_id}/foo-${i}` },
95
      });
96
      ids.add(await persistClient.serverId());
97
      persistClients.push(persistClient);
98
      const { seq } = await persistClient.set({
99
        messageData: messageData(i, { headers: { [i]: i } }),
100
      });
101
      expect(seq).toBe(2);
102
    }
103
    expect(ids.size).toBe(numServers);
104
  });
105

106
  it("cleans up", () => {
107
    for (const client of persistClients) {
108
      client.close();
109
    }
110
    for (const server of persistServers) {
111
      server.close();
112
    }
113
  });
114
});
115

116
afterAll(async () => {
117
  // slight delay so all the sqlites of the persist servers can finish
118
  // writing to disk, so we can delete the temp directory
119
  await delay(100);
120
  await after();
121
});
122

123