Integrate selected external resolver authority fencing

This commit is contained in:
Dotta committed 2026-10-30 18:43:50 -05:00
commit fef4c0ab35
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);
+33 -22
View File
@@ -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) {