mirror of
https://github.com/Termix-SSH/Termix.git
synced 2026-10-30 15:15:24 +00:00
fix: keep concurrent ssh pool connects from overwriting each other (#1619)
This commit is contained in:
1 parent
2fe7530cd4
commit
fb04edc497
2 files changed
+53
-5
No files matched your search
@@ -86,7 +86,6 @@ export class SSHConnectionPool {
|
|||||||
private async createPooledClient(
|
private async createPooledClient(
|
||||||
key: string,
|
key: string,
|
||||||
factory: () => Promise<Client>,
|
factory: () => Promise<Client>,
|
||||||
existing: PooledConnection[],
|
|
||||||
): Promise<Client> {
|
): Promise<Client> {
|
||||||
const generation = this.generations.get(key) || 0;
|
const generation = this.generations.get(key) || 0;
|
||||||
this.pendingConnections.set(
|
this.pendingConnections.set(
|
||||||
@@ -109,8 +108,11 @@ export class SSHConnectionPool {
|
|||||||
inUse: true,
|
inUse: true,
|
||||||
hostKey: key,
|
hostKey: key,
|
||||||
};
|
};
|
||||||
existing.push(pooled);
|
// Re-read after the await: concurrent connects and removals may have
|
||||||
this.connections.set(key, existing);
|
// replaced this key's array in the meantime.
|
||||||
|
const connections = this.connections.get(key) ?? [];
|
||||||
|
connections.push(pooled);
|
||||||
|
this.connections.set(key, connections);
|
||||||
|
|
||||||
client.on("end", () => {
|
client.on("end", () => {
|
||||||
this.removeConnection(key, client);
|
this.removeConnection(key, client);
|
||||||
@@ -187,7 +189,7 @@ export class SSHConnectionPool {
|
|||||||
if (queue.length === 0) this.waiters.delete(key);
|
if (queue.length === 0) this.waiters.delete(key);
|
||||||
else this.waiters.set(key, queue);
|
else this.waiters.set(key, queue);
|
||||||
clearTimeout(waiter.timer);
|
clearTimeout(waiter.timer);
|
||||||
this.createPooledClient(key, waiter.factory, connections)
|
this.createPooledClient(key, waiter.factory)
|
||||||
.then(waiter.resolve)
|
.then(waiter.resolve)
|
||||||
.catch(waiter.reject);
|
.catch(waiter.reject);
|
||||||
return;
|
return;
|
||||||
@@ -225,7 +227,7 @@ export class SSHConnectionPool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
if (this.hasCapacity(key, connections)) {
|
if (this.hasCapacity(key, connections)) {
|
||||||
return this.createPooledClient(key, factory, connections);
|
return this.createPooledClient(key, factory);
|
||||||
}
|
}
|
||||||
|
|
||||||
return this.enqueueWaiter(key, factory);
|
return this.enqueueWaiter(key, factory);
|
||||||
|
|||||||
@@ -42,6 +42,52 @@ describe("SSHConnectionPool", () => {
|
|||||||
pool.destroy();
|
pool.destroy();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("tracks every client from concurrent connects to a new host", async () => {
|
||||||
|
const pool = new SSHConnectionPool();
|
||||||
|
const clients = Array.from({ length: 3 }, deferredClient);
|
||||||
|
let factoryCalls = 0;
|
||||||
|
const factory = vi.fn(() => clients[factoryCalls++].promise);
|
||||||
|
|
||||||
|
const requests = Array.from({ length: 3 }, () =>
|
||||||
|
pool.getConnection("new-host", factory),
|
||||||
|
);
|
||||||
|
for (const pending of clients) pending.resolve(pending.client);
|
||||||
|
const connected = await Promise.all(requests);
|
||||||
|
|
||||||
|
for (const client of connected) pool.releaseConnection("new-host", client);
|
||||||
|
const reused = await Promise.all(
|
||||||
|
Array.from({ length: 3 }, () => pool.getConnection("new-host", factory)),
|
||||||
|
);
|
||||||
|
expect(new Set(reused)).toEqual(new Set(connected));
|
||||||
|
expect(factory).toHaveBeenCalledTimes(3);
|
||||||
|
|
||||||
|
pool.clearKeyConnections("new-host");
|
||||||
|
for (const pending of clients) {
|
||||||
|
expect(pending.client.end).toHaveBeenCalledOnce();
|
||||||
|
}
|
||||||
|
pool.destroy();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("does not resurrect a connection removed while another connects", async () => {
|
||||||
|
const pool = new SSHConnectionPool();
|
||||||
|
const first = deferredClient();
|
||||||
|
const second = deferredClient();
|
||||||
|
first.resolve(first.client);
|
||||||
|
await pool.getConnection("host", () => first.promise);
|
||||||
|
|
||||||
|
const request = pool.getConnection("host", () => second.promise);
|
||||||
|
first.client.emit("close");
|
||||||
|
second.resolve(second.client);
|
||||||
|
await expect(request).resolves.toBe(second.client);
|
||||||
|
|
||||||
|
pool.releaseConnection("host", first.client);
|
||||||
|
pool.releaseConnection("host", second.client);
|
||||||
|
await expect(
|
||||||
|
pool.getConnection("host", () => deferredClient().promise),
|
||||||
|
).resolves.toBe(second.client);
|
||||||
|
pool.destroy();
|
||||||
|
});
|
||||||
|
|
||||||
it("discards a pending connection when its host pool is cleared", async () => {
|
it("discards a pending connection when its host pool is cleared", async () => {
|
||||||
const pool = new SSHConnectionPool();
|
const pool = new SSHConnectionPool();
|
||||||
const pending = deferredClient();
|
const pending = deferredClient();
|
||||||
|
|||||||
Reference in new issue
Block a user