mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 08:22:20 +02:00
perf(slack): overlap endpoint preparation and narrow credential reads (#16593)
Prepare Slack endpoints in pairs with a per-call concurrency limit of two, and resolve only the current bot token for live tool authority. Preserve candidate-order errors, active-sibling settlement, credential rotation and authorization revocation. Verified with real PostgreSQL regression controls, full repository CI/E2E, security checks and current-head Greptile Plus5/5 with no unresolved findings. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
d1463fb79e
commit
3db33c131e
9 files changed
+284
-18
No files matched your search
@@ -44,6 +44,13 @@ signed approvals. Removing a tool profile binding does not get undone by resolvi
|
||||
a retained session. Disconnecting an identity or disabling a connection revokes
|
||||
access for retained runs too.
|
||||
|
||||
Assignment and tool discovery prepare at most two endpoints at a time. Each
|
||||
endpoint still resolves live authority. Preparation waits for the active pair
|
||||
before it returns or fails, keeps candidate order, and stops before the next
|
||||
pair if either check fails. Tool authority resolves only the current bot token;
|
||||
webhook signing and app tokens stay on their existing transport paths. No
|
||||
credential or authorization cache is introduced.
|
||||
|
||||
## Tools and scope upgrades
|
||||
|
||||
The reviewed, strict method/scope/argument matrix is
|
||||
|
||||
@@ -129,6 +129,15 @@ The pinned SDK transport regression is covered in
|
||||
[receipt tests](../../server/src/services/chat-slack-receipts.test.ts); live
|
||||
acknowledgement speed needs a separate provider test.
|
||||
|
||||
Slack assignment and tool discovery bound concurrent endpoint preparation to
|
||||
two. Every endpoint keeps the live task, company, agent and linked-user checks;
|
||||
403 denials omit only that endpoint. Preparation waits for both active checks
|
||||
before it returns an ordered result or propagates an error. Tool authority reads
|
||||
the current bot token without resolving unrelated webhook credentials. See the
|
||||
[real-database preparation controls](../../server/src/__tests__/slack-tool-preparation.test.ts)
|
||||
and [batch failure controls](../../server/src/services/connectors/slack-endpoint-preparation.test.ts).
|
||||
These local controls do not establish live Slack latency or provider access.
|
||||
|
||||
New conversations resolve typed project/workspace defaults from their endpoint
|
||||
and destination under the task-creation transaction. Defaults do not authorize
|
||||
private sources: admission checks the linked user's authority and assigned agent.
|
||||
|
||||
@@ -60,6 +60,14 @@ and removes them when the final reply arrives. The pinned Slack SDK encodes a
|
||||
linear DM with an empty thread timestamp; receipt transport accepts that address
|
||||
while still validating the channel and the source message timestamp.
|
||||
|
||||
For an agent with two linked Slack endpoints, compare assignment preparation and
|
||||
tool discovery in the run trace. Endpoint checks may overlap in pairs. Confirm
|
||||
that both permitted tool sets remain available, then revoke one linked identity
|
||||
and confirm that the next preparation excludes it. The
|
||||
[database regression](../../server/src/__tests__/slack-tool-preparation.test.ts)
|
||||
uses real PostgreSQL with one connection and controlled credential resolution;
|
||||
it does not prove live provider latency or an end-to-end response target.
|
||||
|
||||
### `endpoint-management`
|
||||
|
||||
Automated: [channel integration](../../server/src/__tests__/chat-channels.integration.test.ts) covers shared endpoint/delivery behavior; live credential repair and provider ownership remain manual checks.
|
||||
|
||||
@@ -0,0 +1,152 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import {
|
||||
agents, chatEndpoints, chatExternalPrincipals, chatIdentityLinks, companies,
|
||||
companyMemberships, createDb, heartbeatRuns, issues, toolApplications, toolConnections,
|
||||
} from "@paperclipai/db";
|
||||
import type { ToolCredentialSecretRef } from "@paperclipai/shared";
|
||||
import { createChatCredentials } from "../services/chat-channels/credentials.js";
|
||||
import { createSlackTaskAuthorityRegistration } from "../services/chat-channels/slack/task-authority.js";
|
||||
import { resolveSlackTaskAuthority } from "../services/connectors/slack-authority.js";
|
||||
import { slackAssignedResource } from "../services/connectors/slack.js";
|
||||
import { slackToolsForSession, syncSlackBotTools } from "../services/connectors/slack-catalog.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
import { initializeRunIdentity } from "../services/run-identity.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const describeDatabase = support.supported ? describe : describe.skip;
|
||||
|
||||
function deferred() {
|
||||
let resolve!: () => void;
|
||||
const promise = new Promise<void>(yes => { resolve = yes; });
|
||||
return { promise, resolve };
|
||||
}
|
||||
|
||||
describeDatabase("Slack tool preparation", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("paperclip-slack-preparation-");
|
||||
db = createDb(database.connectionString, { maxConnections: 1 });
|
||||
await instanceSettingsService(db).updateExperimental({ enableChatConnectors: true });
|
||||
});
|
||||
afterAll(async () => { await database?.cleanup(); });
|
||||
|
||||
async function fixture() {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(), userId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Slack preparation", issuePrefix: randomUUID().slice(0, 8).toUpperCase() });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Slack agent" });
|
||||
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "owner" });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Briefing", assigneeAgentId: agentId, responsibleUserId: userId });
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", responsibleUserId: userId, contextSnapshot: { issueId, source: "routine" } });
|
||||
await initializeRunIdentity(db, { companyId, agentId, issueId, runId, responsibleUserId: userId, cause: "routine" });
|
||||
const refs: ToolCredentialSecretRef[] = [];
|
||||
const endpoints: Array<typeof chatEndpoints.$inferSelect> = [];
|
||||
const links: Array<typeof chatIdentityLinks.$inferSelect> = [];
|
||||
for (let i = 0; i < 2; i++) {
|
||||
const [application] = await db.insert(toolApplications).values({ companyId, name: `Slack ${i}`, type: "mcp_http" }).returning();
|
||||
const credentialSecretRefs = ["botToken", "signingSecret", "appToken"].map(key => ({ secretId: randomUUID(), configPath: `credentials.${key}`, versionSelector: "latest" as const, required: true }));
|
||||
refs.push(...credentialSecretRefs);
|
||||
const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: application.id, name: `Slack ${i}`, uid: randomUUID(), transport: "mcp_remote", credentialSecretRefs }).returning();
|
||||
const [endpoint] = await db.insert(chatEndpoints).values({ companyId, connectionId: connection.id, assignedAgentId: agentId, provider: "slack", publicId: randomUUID(), providerAccountId: `T${i}`, status: "active" }).returning();
|
||||
endpoints.push(endpoint);
|
||||
const [principal] = await db.insert(chatExternalPrincipals).values({ companyId, provider: "slack", providerAccountId: `T${i}`, externalId: `U${i}`, isBot: false }).returning();
|
||||
const [link] = await db.insert(chatIdentityLinks).values({ companyId, endpointId: endpoint.id, principalId: principal.id, paperclipUserId: userId, status: "linked" }).returning();
|
||||
links.push(link);
|
||||
await db.transaction(tx => syncSlackBotTools(tx, endpoint, userId, true));
|
||||
}
|
||||
const token = "current-bot-token";
|
||||
const resolveSecretValue = vi.fn(async (_companyId: string, secretId: string) => {
|
||||
const ref = refs.find(ref => ref.secretId === secretId)!;
|
||||
return ref.configPath === "credentials.botToken" ? token : "webhook-secret";
|
||||
});
|
||||
const credentials = createChatCredentials(db, { secrets: { resolveSecretValue } } as unknown as Parameters<typeof createChatCredentials>[1]);
|
||||
const registration = createSlackTaskAuthorityRegistration(db, {
|
||||
resolveCredentials: credentials.resolveCredentials,
|
||||
authorizeInboundWakeup: async () => { throw new Error("This is an ordinary task"); },
|
||||
});
|
||||
return { binding: { companyId, agentId, issueId, runId }, endpoints, links, refs, resolveSecretValue,
|
||||
...registration };
|
||||
}
|
||||
|
||||
const paths = {
|
||||
assignments: async (binding: { companyId: string; agentId: string; runId: string; issueId: string }) =>
|
||||
(await slackAssignedResource(db, binding)).map(resource => resource.id),
|
||||
tools: async (binding: { companyId: string; agentId: string; runId: string; issueId: string }) =>
|
||||
[...new Set((await slackToolsForSession(db, binding)).map(tool => String(tool.providerMetadata?.endpointId)))],
|
||||
};
|
||||
|
||||
it.each(Object.entries(paths))("overlaps two independent endpoint checks for %s", async (_name, prepare) => {
|
||||
const f = await fixture();
|
||||
const gate = deferred();
|
||||
let active = 0, peak = 0;
|
||||
f.resolveSecretValue.mockImplementation(async (_companyId, secretId) => {
|
||||
if (f.refs.find(ref => ref.secretId === secretId)!.configPath === "credentials.botToken") {
|
||||
active++; peak = Math.max(peak, active);
|
||||
await gate.promise;
|
||||
active--;
|
||||
}
|
||||
return "credential";
|
||||
});
|
||||
const pending = prepare(f.binding);
|
||||
try {
|
||||
await vi.waitFor(() => expect(peak).toBe(2), { timeout: 1000, interval: 10 });
|
||||
} finally {
|
||||
gate.resolve();
|
||||
await pending;
|
||||
f.unregisterSlackTaskAuthority();
|
||||
}
|
||||
expect((await pending).sort()).toEqual(f.endpoints.map(endpoint => endpoint.id).sort());
|
||||
});
|
||||
|
||||
it("resolves only current bot credentials and observes rotation without a cache", async () => {
|
||||
const f = await fixture();
|
||||
f.resolveSecretValue.mockImplementation(async (_companyId, secretId) => {
|
||||
const ref = f.refs.find(ref => ref.secretId === secretId)!;
|
||||
if (ref.configPath !== "credentials.botToken") throw new Error("Webhook credential is unavailable");
|
||||
return f.resolveSecretValue.mock.calls.length === 1 ? "old-bot-token" : "rotated-bot-token";
|
||||
});
|
||||
try {
|
||||
const binding = { ...f.binding, endpointId: f.endpoints[0].id };
|
||||
const before = await resolveSlackTaskAuthority(db, binding);
|
||||
const after = await resolveSlackTaskAuthority(db, binding);
|
||||
expect(before.botToken).toBe("old-bot-token");
|
||||
expect(after.botToken).toBe("rotated-bot-token");
|
||||
expect(after.revision).not.toBe(before.revision);
|
||||
expect(f.resolveSecretValue).toHaveBeenCalledTimes(2);
|
||||
} finally { f.unregisterSlackTaskAuthority(); }
|
||||
});
|
||||
|
||||
it("preserves a failed or missing bot credential", async () => {
|
||||
const f = await fixture();
|
||||
const binding = { ...f.binding, endpointId: f.endpoints[0].id };
|
||||
try {
|
||||
const error = new Error("Bot credential is revoked");
|
||||
f.resolveSecretValue.mockRejectedValueOnce(error);
|
||||
await expect(resolveSlackTaskAuthority(db, binding)).rejects.toBe(error);
|
||||
f.resolveSecretValue.mockResolvedValueOnce("");
|
||||
await expect(resolveSlackTaskAuthority(db, binding)).rejects.toMatchObject({ status: 403 });
|
||||
} finally { f.unregisterSlackTaskAuthority(); }
|
||||
});
|
||||
|
||||
it.each(Object.entries(paths))("rechecks identity revocation for %s without touching another company", async (_name, prepare) => {
|
||||
const f = await fixture();
|
||||
const foreign = await fixture();
|
||||
foreign.unregisterSlackTaskAuthority();
|
||||
const registration = createSlackTaskAuthorityRegistration(db, {
|
||||
resolveCredentials: createChatCredentials(db, { secrets: { resolveSecretValue: f.resolveSecretValue } } as unknown as Parameters<typeof createChatCredentials>[1]).resolveCredentials,
|
||||
authorizeInboundWakeup: async () => { throw new Error("This is an ordinary task"); },
|
||||
});
|
||||
try {
|
||||
expect((await prepare(f.binding)).sort()).toEqual(f.endpoints.map(endpoint => endpoint.id).sort());
|
||||
await db.update(chatIdentityLinks).set({ status: "revoked" }).where(eq(chatIdentityLinks.id, f.links[0].id));
|
||||
f.resolveSecretValue.mockClear();
|
||||
expect(await prepare(f.binding)).toEqual([f.endpoints[1].id]);
|
||||
expect(f.resolveSecretValue).toHaveBeenCalledTimes(1);
|
||||
expect(f.resolveSecretValue.mock.calls[0][0]).toBe(f.binding.companyId);
|
||||
} finally { registration.unregisterSlackTaskAuthority(); }
|
||||
});
|
||||
});
|
||||
@@ -1,4 +1,5 @@
|
||||
import type { EndpointRow, DbOrTransaction } from "../runtime-types.js";
|
||||
import type { createChatCredentials } from "../credentials.js";
|
||||
|
||||
import { slackSearchActionToken } from "../../connectors/slack-search-context.js";
|
||||
import { slackAuthorizationRevision } from "../../connectors/slack-revision.js";
|
||||
@@ -45,7 +46,7 @@ type SlackTaskAuthorityRegistrationOptions = {
|
||||
endpoint: EndpointRow;
|
||||
conversation: typeof chatConversations.$inferSelect;
|
||||
}>;
|
||||
resolveCredentials: (endpoint: EndpointRow) => Promise<Record<string, string>>;
|
||||
resolveCredentials: ReturnType<typeof createChatCredentials>["resolveCredentials"];
|
||||
};
|
||||
|
||||
export function createSlackTaskAuthorityRegistration(db: Db, options: SlackTaskAuthorityRegistrationOptions) {
|
||||
@@ -136,7 +137,7 @@ export function createSlackTaskAuthorityRegistration(db: Db, options: SlackTaskA
|
||||
if (!principal) throw forbidden("Slack sender is no longer available");
|
||||
return { ...source, principalId: principal.id, slackUserId: principal.externalId };
|
||||
});
|
||||
const credentials = await measureRunPhase("connectors.slack.credentials", () => resolveCredentials(resolved.endpoint));
|
||||
const credentials = await measureRunPhase("connectors.slack.credentials", () => resolveCredentials(resolved.endpoint, db, ["botToken"]));
|
||||
if (!credentials.botToken) throw forbidden("Slack bot credential is unavailable");
|
||||
return {
|
||||
endpoint: resolved.endpoint, issueId: binding.issueId, conversation: resolved.conversation,
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { prepareSlackEndpoints } from "./slack-endpoint-preparation.js";
|
||||
import { createHash } from "node:crypto";
|
||||
import { measureRunPhase } from "../execution-preparation-timing.js";
|
||||
import { and, eq, isNull, sql } from "drizzle-orm";
|
||||
@@ -271,12 +272,9 @@ export async function slackToolsForSession(
|
||||
session: Parameters<typeof slackToolsForEndpoint>[1],
|
||||
): Promise<ToolGatewayDescriptor[]> {
|
||||
if (!session.agentId || !session.runId || !session.issueId) return [];
|
||||
const tools: ToolGatewayDescriptor[] = [];
|
||||
for (const endpoint of await slackEndpointCandidates(db, {
|
||||
const endpoints = await slackEndpointCandidates(db, {
|
||||
companyId: session.companyId,
|
||||
agentId: session.agentId,
|
||||
})) {
|
||||
tools.push(...(await slackToolsForEndpoint(db, session, endpoint.id)));
|
||||
}
|
||||
return tools;
|
||||
});
|
||||
return prepareSlackEndpoints(endpoints, endpoint => slackToolsForEndpoint(db, session, endpoint.id));
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { prepareSlackEndpoints } from "./slack-endpoint-preparation.js";
|
||||
|
||||
function deferred<T>() {
|
||||
let resolve!: (value: T) => void;
|
||||
let reject!: (error: Error) => void;
|
||||
const promise = new Promise<T>((yes, no) => { resolve = yes; reject = no; });
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
const endpoints = Array.from({ length: 5 }, (_, i) => ({ id: String(i), connectionId: `connection-${i}` }));
|
||||
|
||||
describe("Slack endpoint preparation batches", () => {
|
||||
it("does not start work without candidates", async () => {
|
||||
const resolve = vi.fn();
|
||||
expect(await prepareSlackEndpoints([], resolve)).toEqual([]);
|
||||
expect(resolve).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("keeps at most two checks active and preserves candidate order", async () => {
|
||||
const pending = endpoints.map(() => deferred<string[]>());
|
||||
const resolve = vi.fn((endpoint: typeof endpoints[number]) => pending[Number(endpoint.id)].promise);
|
||||
const prepared = prepareSlackEndpoints(endpoints, resolve);
|
||||
expect(resolve.mock.calls.map(([endpoint]) => endpoint.id)).toEqual(["0", "1"]);
|
||||
pending[1].resolve(["second"]);
|
||||
await Promise.resolve();
|
||||
expect(resolve).toHaveBeenCalledTimes(2);
|
||||
pending[0].resolve(["first"]);
|
||||
await vi.waitFor(() => expect(resolve).toHaveBeenCalledTimes(4));
|
||||
pending[3].resolve(["fourth"]);
|
||||
pending[2].resolve([]);
|
||||
await vi.waitFor(() => expect(resolve).toHaveBeenCalledTimes(5));
|
||||
pending[4].resolve(["fifth"]);
|
||||
expect(await prepared).toEqual(["first", "second", "fourth", "fifth"]);
|
||||
});
|
||||
|
||||
it("settles the active batch before rejecting and does not start later candidates", async () => {
|
||||
const gate = deferred<string[]>();
|
||||
const error = new Error("Endpoint unavailable");
|
||||
const resolve = vi.fn((endpoint: typeof endpoints[number]) => endpoint.id === "0" ? Promise.reject(error) : gate.promise);
|
||||
let finished = false;
|
||||
const prepared = prepareSlackEndpoints(endpoints, resolve);
|
||||
const checked = expect(prepared).rejects.toBe(error).finally(() => { finished = true; });
|
||||
await Promise.resolve();
|
||||
expect(finished).toBe(false);
|
||||
expect(resolve).toHaveBeenCalledTimes(2);
|
||||
gate.resolve([]);
|
||||
await checked;
|
||||
expect(resolve).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("keeps candidate-order failure precedence even if the second fails first", async () => {
|
||||
const first = deferred<string[]>();
|
||||
const firstError = new Error("First candidate failed");
|
||||
const secondError = new Error("Second candidate failed");
|
||||
const resolve = vi.fn((endpoint: typeof endpoints[number]) => endpoint.id === "0" ? first.promise : Promise.reject(secondError));
|
||||
const checked = expect(prepareSlackEndpoints(endpoints, resolve)).rejects.toBe(firstError);
|
||||
await Promise.resolve();
|
||||
first.reject(firstError);
|
||||
await checked;
|
||||
expect(resolve).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("settles a sibling after a synchronous resolver failure", async () => {
|
||||
const gate = deferred<string[]>();
|
||||
const error = new Error("Synchronous failure");
|
||||
const resolve = vi.fn((endpoint: typeof endpoints[number]) => {
|
||||
if (endpoint.id === "0") throw error;
|
||||
return gate.promise;
|
||||
});
|
||||
const checked = expect(prepareSlackEndpoints(endpoints, resolve)).rejects.toBe(error);
|
||||
expect(resolve).toHaveBeenCalledTimes(2);
|
||||
gate.resolve([]);
|
||||
await checked;
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,20 @@
|
||||
type SlackEndpoint = { id: string; connectionId: string };
|
||||
|
||||
export async function prepareSlackEndpoints<T>(
|
||||
endpoints: readonly SlackEndpoint[],
|
||||
resolve: (endpoint: SlackEndpoint) => Promise<T[]>,
|
||||
): Promise<T[]> {
|
||||
const contributions: T[] = [];
|
||||
for (let offset = 0; offset < endpoints.length; offset += 2) {
|
||||
const results = await Promise.allSettled(
|
||||
endpoints.slice(offset, offset + 2).map(async endpoint => resolve(endpoint)),
|
||||
);
|
||||
for (const result of results) {
|
||||
if (result.status === "rejected") throw result.reason;
|
||||
}
|
||||
for (const result of results) {
|
||||
if (result.status === "fulfilled") contributions.push(...result.value);
|
||||
}
|
||||
}
|
||||
return contributions;
|
||||
}
|
||||
@@ -6,6 +6,7 @@ import { slackBotCatalogIsCurrent, syncSlackBotTools } from "./slack-catalog.js"
|
||||
import { SLACK_TOOLS } from "@paperclipai/shared";
|
||||
import { measureRunPhase } from "../execution-preparation-timing.js";
|
||||
import { measureRunTransaction } from "../execution-transaction-timing.js";
|
||||
import { prepareSlackEndpoints } from "./slack-endpoint-preparation.js";
|
||||
import {
|
||||
toolProfiles,
|
||||
toolProfileBindings,
|
||||
@@ -104,16 +105,11 @@ export async function slackAssignedResource(
|
||||
db: Db,
|
||||
binding: Partial<SlackTaskBinding> & { companyId: string; agentId: string },
|
||||
) {
|
||||
const resources = [];
|
||||
for (const endpoint of await measureRunPhase("connectors.slack.candidates", () => slackEndpointCandidates(db, binding))) {
|
||||
resources.push(
|
||||
...(await slackAssignedEndpoint(db, {
|
||||
...binding,
|
||||
endpointId: endpoint.id,
|
||||
}, endpoint)),
|
||||
);
|
||||
}
|
||||
return resources;
|
||||
const endpoints = await measureRunPhase("connectors.slack.candidates", () => slackEndpointCandidates(db, binding));
|
||||
return prepareSlackEndpoints(endpoints, endpoint => slackAssignedEndpoint(db, {
|
||||
...binding,
|
||||
endpointId: endpoint.id,
|
||||
}, endpoint));
|
||||
}
|
||||
|
||||
const messageView = (message: SlackObject, team: string, channel: string) => ({
|
||||
|
||||
Reference in new issue
Block a user