mirror of
https://github.com/block/buzz.git
synced 2026-11-01 06:51:17 +00:00
fix(projects): bound channel request queue
Co-authored-by: Wrench <0eabe6ea5758c1e4c5b68cea4ac42b32c479072883cb28da8110b4a47c32b9a1@buzz.block.builderlab.xyz> Signed-off-by: Wrench <0eabe6ea5758c1e4c5b68cea4ac42b32c479072883cb28da8110b4a47c32b9a1@buzz.block.builderlab.xyz>
This commit is contained in:
@@ -5,6 +5,7 @@ import {
|
||||
advanceProjectChannelRequestQueue,
|
||||
createProjectChannelRequestQueue,
|
||||
enqueueProjectChannelRequest,
|
||||
MAX_PENDING_PROJECT_CHANNEL_REQUESTS,
|
||||
} from "./projectChannelRequestQueue.ts";
|
||||
|
||||
function candidate(requestId) {
|
||||
@@ -19,10 +20,63 @@ test("accepted requests advance in order while duplicate active requests stay su
|
||||
const first = candidate("request-a");
|
||||
const second = candidate("request-b");
|
||||
|
||||
assert.equal(enqueueProjectChannelRequest(queue, first), first);
|
||||
assert.equal(enqueueProjectChannelRequest(queue, second), null);
|
||||
assert.equal(enqueueProjectChannelRequest(queue, first), null);
|
||||
assert.deepEqual(enqueueProjectChannelRequest(queue, first), {
|
||||
status: "show",
|
||||
candidate: first,
|
||||
});
|
||||
assert.deepEqual(enqueueProjectChannelRequest(queue, second), {
|
||||
status: "queued",
|
||||
});
|
||||
assert.deepEqual(enqueueProjectChannelRequest(queue, first), {
|
||||
status: "duplicate",
|
||||
});
|
||||
assert.equal(advanceProjectChannelRequestQueue(queue), second);
|
||||
assert.equal(enqueueProjectChannelRequest(queue, first), null);
|
||||
assert.deepEqual(enqueueProjectChannelRequest(queue, first), {
|
||||
status: "duplicate",
|
||||
});
|
||||
assert.equal(advanceProjectChannelRequestQueue(queue), null);
|
||||
});
|
||||
|
||||
test("accepted queue drops newest overflow while preserving FIFO and retryability", () => {
|
||||
const queue = createProjectChannelRequestQueue();
|
||||
const first = candidate("request-0");
|
||||
assert.equal(enqueueProjectChannelRequest(queue, first).status, "show");
|
||||
|
||||
const pending = Array.from(
|
||||
{ length: MAX_PENDING_PROJECT_CHANNEL_REQUESTS },
|
||||
(_, index) => candidate(`request-${index + 1}`),
|
||||
);
|
||||
for (const request of pending) {
|
||||
assert.equal(enqueueProjectChannelRequest(queue, request).status, "queued");
|
||||
}
|
||||
|
||||
const overflow = candidate("request-overflow");
|
||||
assert.deepEqual(enqueueProjectChannelRequest(queue, overflow), {
|
||||
status: "overflow",
|
||||
});
|
||||
assert.equal(queue.pending.length, MAX_PENDING_PROJECT_CHANNEL_REQUESTS);
|
||||
assert.equal(queue.seenRequestIds.has(overflow.request.requestId), false);
|
||||
|
||||
for (const request of pending) {
|
||||
assert.equal(advanceProjectChannelRequestQueue(queue), request);
|
||||
}
|
||||
assert.equal(advanceProjectChannelRequestQueue(queue), null);
|
||||
assert.deepEqual(enqueueProjectChannelRequest(queue, overflow), {
|
||||
status: "show",
|
||||
candidate: overflow,
|
||||
});
|
||||
});
|
||||
|
||||
test("dedup history stays bounded without forgetting active or pending requests", () => {
|
||||
const queue = createProjectChannelRequestQueue();
|
||||
for (let index = 0; index < 500; index += 1) {
|
||||
const request = candidate(`request-${index}`);
|
||||
assert.equal(enqueueProjectChannelRequest(queue, request).status, "show");
|
||||
assert.deepEqual(enqueueProjectChannelRequest(queue, request), {
|
||||
status: "duplicate",
|
||||
});
|
||||
assert.equal(advanceProjectChannelRequestQueue(queue), null);
|
||||
}
|
||||
|
||||
assert.ok(queue.seenRequestIds.size <= 201);
|
||||
});
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
import type { ProjectChannelRequest } from "@/features/projects/projectChannelRequest";
|
||||
|
||||
export const MAX_PENDING_PROJECT_CHANNEL_REQUESTS = 100;
|
||||
const MAX_SEEN_PROJECT_CHANNEL_REQUEST_IDS =
|
||||
MAX_PENDING_PROJECT_CHANNEL_REQUESTS * 2 + 1;
|
||||
|
||||
export type AcceptedProjectChannelRequest = {
|
||||
agentPubkey: string;
|
||||
request: ProjectChannelRequest;
|
||||
@@ -11,6 +15,10 @@ export type ProjectChannelRequestQueue = {
|
||||
seenRequestIds: Set<string>;
|
||||
};
|
||||
|
||||
export type EnqueueProjectChannelRequestResult =
|
||||
| { status: "show"; candidate: AcceptedProjectChannelRequest }
|
||||
| { status: "queued" | "duplicate" | "overflow" };
|
||||
|
||||
export function createProjectChannelRequestQueue(): ProjectChannelRequestQueue {
|
||||
return {
|
||||
activeRequestId: null,
|
||||
@@ -22,18 +30,29 @@ export function createProjectChannelRequestQueue(): ProjectChannelRequestQueue {
|
||||
export function enqueueProjectChannelRequest(
|
||||
queue: ProjectChannelRequestQueue,
|
||||
candidate: AcceptedProjectChannelRequest,
|
||||
): AcceptedProjectChannelRequest | null {
|
||||
): EnqueueProjectChannelRequestResult {
|
||||
const requestId = candidate.request.requestId;
|
||||
if (queue.seenRequestIds.has(requestId)) return null;
|
||||
if (queue.seenRequestIds.has(requestId)) return { status: "duplicate" };
|
||||
|
||||
if (
|
||||
queue.activeRequestId !== null &&
|
||||
queue.pending.length >= MAX_PENDING_PROJECT_CHANNEL_REQUESTS
|
||||
) {
|
||||
// Keep the requests already visible to the owner and drop the newest one.
|
||||
// Do not mark it seen: a later retry may be accepted after space opens.
|
||||
return { status: "overflow" };
|
||||
}
|
||||
|
||||
queue.seenRequestIds.add(requestId);
|
||||
pruneSeenRequestIds(queue);
|
||||
|
||||
if (queue.activeRequestId === null) {
|
||||
queue.activeRequestId = requestId;
|
||||
return candidate;
|
||||
return { status: "show", candidate };
|
||||
}
|
||||
|
||||
queue.pending.push(candidate);
|
||||
return null;
|
||||
return { status: "queued" };
|
||||
}
|
||||
|
||||
export function advanceProjectChannelRequestQueue(
|
||||
@@ -41,5 +60,22 @@ export function advanceProjectChannelRequestQueue(
|
||||
): AcceptedProjectChannelRequest | null {
|
||||
const next = queue.pending.shift() ?? null;
|
||||
queue.activeRequestId = next?.request.requestId ?? null;
|
||||
pruneSeenRequestIds(queue);
|
||||
return next;
|
||||
}
|
||||
|
||||
function pruneSeenRequestIds(queue: ProjectChannelRequestQueue) {
|
||||
if (queue.seenRequestIds.size <= MAX_SEEN_PROJECT_CHANNEL_REQUEST_IDS) return;
|
||||
|
||||
const pendingIds = new Set(
|
||||
queue.pending.map((candidate) => candidate.request.requestId),
|
||||
);
|
||||
for (const requestId of queue.seenRequestIds) {
|
||||
if (requestId !== queue.activeRequestId && !pendingIds.has(requestId)) {
|
||||
queue.seenRequestIds.delete(requestId);
|
||||
if (queue.seenRequestIds.size <= MAX_SEEN_PROJECT_CHANNEL_REQUEST_IDS) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -51,11 +51,11 @@ export function useProjectChannelRequests() {
|
||||
) {
|
||||
return;
|
||||
}
|
||||
const accepted = enqueueProjectChannelRequest(acceptedRequests.current, {
|
||||
const result = enqueueProjectChannelRequest(acceptedRequests.current, {
|
||||
agentPubkey,
|
||||
request: next,
|
||||
});
|
||||
if (accepted) showRequest(accepted);
|
||||
if (result.status === "show") showRequest(result.candidate);
|
||||
},
|
||||
);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user