mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-03 01:27:09 +01:00
fix(external-objects): fence selected plugin results
Revalidate the exact selected plugin before result persistence even when an allowed fallback still supports the object. Preserve due debt and token ownership so a fresh allowed resolver can proceed. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
d199b3e0a6
commit
594a4ee657
3 files changed
+96
-23
No files matched your search
@@ -18,6 +18,14 @@ ready plugin with `external.objects.read`, a matching provider/object declaratio
|
||||
and no company setting that disables it. Built-in resolvers remain available
|
||||
independently of optional plugin support.
|
||||
|
||||
A plugin result carries its selected plugin ID inside the host owner. Persistence
|
||||
rechecks that exact plugin's readiness, read capability, object declaration and
|
||||
same-company enabled setting in the result update. Another available resolver
|
||||
cannot authorize an obsolete plugin response. A rejected response releases only
|
||||
its own claim; the due row remains for a fresh allowed resolver. Plugin RPC
|
||||
payloads and public result shapes do not contain this host selection context.
|
||||
|
||||
|
||||
Terminal objects are passive. The existing exception remains: GitHub
|
||||
`not_found` snapshots without `data.probeVerified=true` need a repository-visibility
|
||||
probe before their terminal verdict is definitive. A null deadline retains its
|
||||
@@ -94,7 +102,7 @@ reactivation in the tested process; it does not establish Cloud alarm coverage,
|
||||
host/database suspension or deployed savings.
|
||||
|
||||
|
||||
## Host cancellation and explicit legacy entry points
|
||||
## Host cancellation and registered ownership
|
||||
|
||||
The resolver contract accepts an optional host `AbortSignal`. The worker passes
|
||||
its coordinator signal through refresh to the resolver. This reuses the narrow
|
||||
|
||||
@@ -257,6 +257,60 @@ const active: ExternalObjectResolveResult = { ok: true, snapshot: {
|
||||
expect(calls).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it.each(["company setting", "status", "uninstall", "manifest", "other company"] as const)(
|
||||
"revalidates the selected plugin after %s changes even while the built-in fallback remains available", async change => {
|
||||
clock(); const registry = pluginRegistryService(db);
|
||||
const manifest: PaperclipPluginManifestV1 = {
|
||||
id: `paperclip.selected-${randomUUID()}`, apiVersion: 1, version: "1.0.0", displayName: "Selected provider",
|
||||
description: "Selected resolver fixture", author: "Paperclip", categories: ["connector"],
|
||||
capabilities: ["external.objects.detect", "external.objects.read"], entrypoints: { worker: "worker.js" },
|
||||
objectReferences: [{ providerKey: "github", displayName: "GitHub", objectTypes: ["pull_request"], urlPatterns: ["https://github.com/:owner/:repo/pull/:id"] }],
|
||||
};
|
||||
const plugin = await registry.install({ packageName: "selected-provider-fixture" }, manifest);
|
||||
await registry.updateStatus(plugin!.id, { status: "ready" });
|
||||
let finishPlugin!: (result: ExternalObjectResolveResult) => void;
|
||||
const pluginResult = new Promise<ExternalObjectResolveResult>(resolve => { finishPlugin = resolve; });
|
||||
let selected = false;
|
||||
const calls = vi.fn(async (_id: string, method: string) => {
|
||||
if (method === "detectExternalObjects") return { detections: [] };
|
||||
selected = true; return pluginResult;
|
||||
});
|
||||
let finishFallback!: () => void;
|
||||
const fallbackResult = new Promise<void>(resolve => { finishFallback = resolve; });
|
||||
const fallback = vi.fn(async () => {
|
||||
await fallbackResult;
|
||||
return new Response(JSON.stringify({ state: "closed", merged: true, title: "Allowed fallback" }), { status: 200 });
|
||||
});
|
||||
const svc = externalObjectService(db, { pluginWorkerManager: { call: calls } as unknown as PluginWorkerManager,
|
||||
github: { fetch: fallback, credentialProvider: () => ({ token: "simulated-company-token", source: "company_secret" }) } });
|
||||
try {
|
||||
await start(svc); const f = await fixture();
|
||||
await db.update(issues).set({ description: "https://github.com/acme/app/pull/42" }).where(eq(issues.id, f.issueId));
|
||||
await svc.syncIssue(f.issueId); await settle(() => selected);
|
||||
if (change === "company setting") await registry.upsertCompanySettings(plugin!.id, f.companyId, { enabled: false, settingsJson: {} });
|
||||
else if (change === "status") await registry.updateStatus(plugin!.id, { status: "error" });
|
||||
else if (change === "uninstall") await registry.uninstall(plugin!.id, true);
|
||||
else if (change === "manifest") await registry.update(plugin!.id, { manifest: { ...manifest, capabilities: [], objectReferences: [] } });
|
||||
else {
|
||||
const other = await fixture();
|
||||
await registry.upsertCompanySettings(plugin!.id, other.companyId, { enabled: false, settingsJson: {} });
|
||||
}
|
||||
finishPlugin({ ok: true, snapshot: { statusKey: "plugin_result", statusCategory: "succeeded", statusTone: "success", isTerminal: true } });
|
||||
if (change === "other company") {
|
||||
await settle(async () => (await row(f.companyId))?.statusKey === "plugin_result"); await quiet();
|
||||
expect(fallback).not.toHaveBeenCalled();
|
||||
} else {
|
||||
await settle(async () => fallback.mock.calls.length > 0 || Boolean((await row(f.companyId))?.lastResolvedAt));
|
||||
const beforeFallback = await row(f.companyId);
|
||||
expect(beforeFallback.isTerminal).toBe(false); expect(beforeFallback.statusKey).toBeNull(); expect(beforeFallback.lastResolvedAt).toBeNull();
|
||||
expect(fallback).toHaveBeenCalledOnce();
|
||||
finishFallback();
|
||||
await settle(async () => (await row(f.companyId))?.statusKey === "merged"); await quiet();
|
||||
expect((await row(f.companyId)).displayTitle).toBe("acme/app#42: Allowed fallback");
|
||||
}
|
||||
} finally { finishPlugin(terminal); finishFallback(); }
|
||||
});
|
||||
|
||||
it("reactivates refresh after ordinary budget pause and budget raise", async () => {
|
||||
clock(); const svc = service(); provider.mockResolvedValue(active); await start(svc);
|
||||
const f = await fixture(); await svc.syncIssue(f.issueId);
|
||||
|
||||
@@ -346,7 +346,7 @@ async function resolveViaPluginProvider(
|
||||
db: Db,
|
||||
pluginWorkerManager: PluginWorkerManager | undefined,
|
||||
object: ExternalObjectRecord,
|
||||
): Promise<PluginExternalObjectResolveResult | null> {
|
||||
): Promise<{ pluginId: string; result: PluginExternalObjectResolveResult } | null> {
|
||||
if (!pluginWorkerManager) return null;
|
||||
const providers = await readyObjectReferencePlugins(db, object.companyId);
|
||||
for (const provider of providers) {
|
||||
@@ -354,23 +354,27 @@ async function resolveViaPluginProvider(
|
||||
if (!manifest.capabilities.includes("external.objects.read")) continue;
|
||||
if (!manifestProvidesObject(manifest, object)) continue;
|
||||
try {
|
||||
return await pluginWorkerManager.call(provider.id, "resolveExternalObject", {
|
||||
const result = await pluginWorkerManager.call(provider.id, "resolveExternalObject", {
|
||||
companyId: object.companyId,
|
||||
providerKey: object.providerKey,
|
||||
objectType: object.objectType,
|
||||
externalId: object.externalId,
|
||||
object: objectSnapshot(object),
|
||||
});
|
||||
return { pluginId: provider.id, result };
|
||||
} catch (err) {
|
||||
logger.warn(
|
||||
{ err, pluginId: provider.id, pluginKey: provider.pluginKey, objectId: object.id },
|
||||
"plugin external object resolver failed",
|
||||
);
|
||||
return {
|
||||
ok: false,
|
||||
liveness: "unreachable",
|
||||
errorCode: "plugin_resolver_failed",
|
||||
errorMessage: err instanceof Error ? err.message : String(err),
|
||||
pluginId: provider.id,
|
||||
result: {
|
||||
ok: false,
|
||||
liveness: "unreachable",
|
||||
errorCode: "plugin_resolver_failed",
|
||||
errorMessage: err instanceof Error ? err.message : String(err),
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -402,12 +406,9 @@ export function externalObjectService(
|
||||
];
|
||||
const resolverRegistry = createExternalObjectResolverRegistry(resolvers);
|
||||
|
||||
function automaticRefreshWhere(database: Db = db) {
|
||||
const supported = resolvers.map(resolver => and(
|
||||
eq(externalObjects.providerKey, resolver.providerKey),
|
||||
resolver.objectType ? eq(externalObjects.objectType, resolver.objectType) : undefined,
|
||||
)!);
|
||||
if (opts.pluginWorkerManager) supported.push(exists(database.select({ id: plugins.id }).from(plugins).where(and(
|
||||
function pluginRefreshWhere(database: Db, selectedPluginId?: string) {
|
||||
return exists(database.select({ id: plugins.id }).from(plugins).where(and(
|
||||
selectedPluginId ? eq(plugins.id, selectedPluginId) : undefined,
|
||||
eq(plugins.status, "ready"),
|
||||
sql`${plugins.manifestJson}->'capabilities' ? 'external.objects.read'`,
|
||||
sql`exists (select 1 from jsonb_array_elements(coalesce(${plugins.manifestJson}->'objectReferences', '[]'::jsonb)) as reference
|
||||
@@ -418,7 +419,15 @@ export function externalObjectService(
|
||||
eq(pluginCompanySettings.companyId, externalObjects.companyId),
|
||||
eq(pluginCompanySettings.enabled, false),
|
||||
)))),
|
||||
))));
|
||||
)));
|
||||
}
|
||||
|
||||
function automaticRefreshWhere(database: Db = db) {
|
||||
const supported = resolvers.map(resolver => and(
|
||||
eq(externalObjects.providerKey, resolver.providerKey),
|
||||
resolver.objectType ? eq(externalObjects.objectType, resolver.objectType) : undefined,
|
||||
)!);
|
||||
if (opts.pluginWorkerManager) supported.push(pluginRefreshWhere(database));
|
||||
return and(
|
||||
refreshableObjectWhere(), isNotNull(externalObjects.nextRefreshAt),
|
||||
supported.length ? or(...supported) : sql`false`,
|
||||
@@ -863,12 +872,14 @@ export function externalObjectService(
|
||||
}
|
||||
|
||||
async function persistRefresh(object: ExternalObjectRecord, token: string,
|
||||
patch: Partial<typeof externalObjects.$inferInsert>, scheduled = false) {
|
||||
patch: Partial<typeof externalObjects.$inferInsert>, scheduled = false, selectedPluginId?: string) {
|
||||
return db.transaction(async tx => {
|
||||
await notifyExternalObjectRefresh(tx, object.companyId, object.id);
|
||||
const rows = await tx.update(externalObjects).set(patch)
|
||||
.where(and(refreshOwnerWhere(object, token), scheduled ? automaticRefreshWhere(tx as unknown as Db) : undefined)).returning();
|
||||
if (scheduled && rows.length === 0) await tx.update(externalObjects)
|
||||
.where(and(refreshOwnerWhere(object, token),
|
||||
scheduled ? automaticRefreshWhere(tx as unknown as Db) : undefined,
|
||||
selectedPluginId ? pluginRefreshWhere(tx as unknown as Db, selectedPluginId) : undefined)).returning();
|
||||
if ((scheduled || selectedPluginId) && rows.length === 0) await tx.update(externalObjects)
|
||||
.set({ refreshStartedAt: null, refreshToken: null }).where(refreshOwnerWhere(object, token));
|
||||
return rows;
|
||||
});
|
||||
@@ -881,9 +892,9 @@ export function externalObjectService(
|
||||
refreshToken: string,
|
||||
) {
|
||||
input.signal?.throwIfAborted();
|
||||
const pluginResult = await observeRefreshCancellation(resolveViaPluginProvider(db, opts.pluginWorkerManager, object), input.signal);
|
||||
const resolver = pluginResult ? null : resolverRegistry.find(object);
|
||||
if (!pluginResult && !resolver) {
|
||||
const pluginResolution = await observeRefreshCancellation(resolveViaPluginProvider(db, opts.pluginWorkerManager, object), input.signal);
|
||||
const resolver = pluginResolution ? null : resolverRegistry.find(object);
|
||||
if (!pluginResolution && !resolver) {
|
||||
const [updated] = await persistRefresh(object, refreshToken, {
|
||||
liveness: object.isTerminal ? object.liveness : visibleLiveness(object, now) === "fresh" ? "stale" : object.liveness,
|
||||
nextRefreshAt: addSeconds(now, DEFAULT_RETRY_AFTER_SECONDS),
|
||||
@@ -895,7 +906,7 @@ export function externalObjectService(
|
||||
return { object: toObjectPayload(updated ?? object, now), refreshed: false, reason: "no_resolver" as const };
|
||||
}
|
||||
|
||||
const result = pluginResult ?? await observeRefreshCancellation(
|
||||
const result = pluginResolution?.result ?? await observeRefreshCancellation(
|
||||
resolver!.resolve({ companyId: object.companyId, object, signal: input.signal }), input.signal);
|
||||
input.signal?.throwIfAborted();
|
||||
if (!result.ok) {
|
||||
@@ -913,7 +924,7 @@ export function externalObjectService(
|
||||
refreshStartedAt: null,
|
||||
refreshToken: null,
|
||||
updatedAt: now,
|
||||
}, input.scheduled);
|
||||
}, input.scheduled, pluginResolution?.pluginId);
|
||||
if (!updated) return refreshSupersededResult(object, now);
|
||||
publishLiveEvent({
|
||||
companyId: object.companyId,
|
||||
@@ -950,7 +961,7 @@ export function externalObjectService(
|
||||
const [updated] = await persistRefresh(object, refreshToken, {
|
||||
...patch,
|
||||
lastChangedAt: objectChanged(object, { ...object, ...patch }) ? now : object.lastChangedAt,
|
||||
}, input.scheduled);
|
||||
}, input.scheduled, pluginResolution?.pluginId);
|
||||
if (!updated) return refreshSupersededResult(object, now);
|
||||
const next = updated ?? object;
|
||||
if (objectChanged(object, next) && input.actor) {
|
||||
|
||||
Reference in new issue
Block a user