History

Log whether a preserved-catalog attempt recorded, and tighten the lock and atomicity notes

Rhys Sullivan committed 2 months ago

Commit 2fb74bd1 parent e48fe133 files · +30 −7Browse files

Unified · Split

3 files
  1. packages/core/sdk/src/executor.ts +22 −5
  2. packages/core/sdk/src/plugin.ts +6 −1
  3. packages/core/sdk/src/tools-sync-scope.test.ts +2 −1
packages/core/sdk/src/executor.ts+22−5
@@ -2871,8 +2871,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 2870 unmodified lines
2871 // column only if this exact value is still in it, so a mark written2871 // column only if this exact value is still in it, so a mark written
2872 // while we were dialing carries a different token, survives the clear,2872 // while we were dialing carries a different token, survives the clear,
2873 // and re-lists the connection on the next read.2873 // and re-lists the connection on the next read.
2874 const observedStaleToken =2874 const observedStaleToken = existingRow?.tools_stale_token ?? null;
2875 existingRow?.tools_stale_token == null ? null : String(existingRow.tools_stale_token);
2876 2875
2877 // Does this attempt still own the connection? A read-path refresh takes2876 // Does this attempt still own the connection? A read-path refresh takes
2878 // a lease before dialing, and a slow one can lose it: another reader2877 // a lease before dialing, and a slow one can lose it: another reader
@@ -2958,6 +2957,13 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 79 unmodified lines
2958 * drift mark this listing answered — all three together, or none of2957 * drift mark this listing answered — all three together, or none of
2959 * them. Returns false when the attempt lost its lease.2958 * them. Returns false when the attempt lost its lease.
2960 *2959 *
2960 * That atomicity is dialect-conditional: on D1
2961 * (`interactiveTransactions: false`) there is no transaction, the
2962 * statements auto-commit one by one, and an interrupted run can leave a
2963 * prefix of them applied. It still fails CLOSED, because the drift clear
2964 * is the LAST statement: any prefix short of the whole leaves the token
2965 * in place, so the connection stays drifted and re-lists.
2966 *
2961 * Already holds the persist permit, which is not reentrant: nothing in2967 * Already holds the persist permit, which is not reentrant: nothing in
2962 * here may take it again.2968 * here may take it again.
2963 *2969 *
@@ -3075,11 +3081,16 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 111 unmodified lines
3075 // Reporting `incomplete` there would attribute a verdict to a row the3081 // Reporting `incomplete` there would attribute a verdict to a row the
3076 // winning attempt owns.3082 // winning attempt owns.
3077 const recorded = yield* writeSyncOutcome(failureSet(kind, reason));3083 const recorded = yield* writeSyncOutcome(failureSet(kind, reason));
3084 // `recorded` rides along because the warning is emitted either way: a
3085 // lost lease preserved the catalog too, but wrote no ladder entry. Without
3086 // the field, a rate derived from these logs would count attempts the
3087 // `incomplete` counter never saw.
3078 yield* Effect.logWarning("executor tool sync preserved catalog", {3088 yield* Effect.logWarning("executor tool sync preserved catalog", {
3079 reason,3089 reason,
3080 errorKind: kind ?? "unclassified",3090 errorKind: kind ?? "unclassified",
3081 integration: String(ref.integration),3091 integration: String(ref.integration),
3082 connection: String(ref.name),3092 connection: String(ref.name),
3093 recorded,
3083 });3094 });
3084 return yield* persistedTools(recorded ? "incomplete" : "lost_claim");3095 return yield* persistedTools(recorded ? "incomplete" : "lost_claim");
3085 }3096 }
@@ -3098,11 +3109,14 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 12 unmodified lines
3098 // the plugin said nothing about why. Unclassified, so it backs off3109 // the plugin said nothing about why. Unclassified, so it backs off
3099 // but never parks.3110 // but never parks.
3100 const recorded = yield* writeSyncOutcome(failureSet(null, reason));3111 const recorded = yield* writeSyncOutcome(failureSet(null, reason));
3112 // `recorded` for the same reason as above: the warning fires even when
3113 // the lease was lost and nothing was written.
3101 yield* Effect.logWarning("executor tool sync preserved nonzero catalog", {3114 yield* Effect.logWarning("executor tool sync preserved nonzero catalog", {
3102 reason,3115 reason,
3103 integration: String(ref.integration),3116 integration: String(ref.integration),
3104 connection: String(ref.name),3117 connection: String(ref.name),
3105 existingToolCount: keptRows.length,3118 existingToolCount: keptRows.length,
3119 recorded,
3106 });3120 });
3107 // Re-read on a lost lease: `keptRows` predates the write, and the3121 // Re-read on a lost lease: `keptRows` predates the write, and the
3108 // attempt that owns the connection may have replaced the catalog3122 // attempt that owns the connection may have replaced the catalog
@@ -3946,10 +3960,13 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea · 837 unmodified lines
3946 // with a new one is precisely how this signal outlives the listing that3960 // with a new one is precisely how this signal outlives the listing that
3947 // could not have seen it.3961 // could not have seen it.
3948 //3962 //
3949 // Locked: this is a lifecycle write like every other, and it can run while3963 // Locked: the permit serializes this write against the catalog-persist
3950 // a read's refresh fan-out holds an open BEGIN on the shared SQLite3964 // transactions specifically — `replaceCatalog` and `writeSyncOutcome`, the
3951 // connection — where a bare UPDATE is enrolled in that transaction and lost3965 // only permit holders that open a BEGIN. On the shared SQLite connection a
3966 // bare UPDATE issued while one of those is open is enrolled in it, and lost
3952 // with it if it rolls back. Losing THIS write loses a drift signal outright.3967 // with it if it rolls back. Losing THIS write loses a drift signal outright.
3968 // The claim reaches no further: transactions in this file that do not take
3969 // the permit are not serialized against this write.
3953 const connectionsMarkToolsStale = (ref: ConnectionRef): Effect.Effect<void, StorageFailure> =>3970 const connectionsMarkToolsStale = (ref: ConnectionRef): Effect.Effect<void, StorageFailure> =>
3954 withCatalogPersistLock(3971 withCatalogPersistLock(
3955 core.updateMany("connection", {3972 core.updateMany("connection", {
packages/core/sdk/src/plugin.ts+6−1
@@ -242,7 +242,12 @@ export interface PluginCtx<TStore = unknown> { · 241 unmodified lines
242 * Safe to call at any point, including while a listing for the same242 * Safe to call at any point, including while a listing for the same
243 * connection is in flight: the mark is recorded as a fresh token, and a243 * connection is in flight: the mark is recorded as a fresh token, and a
244 * listing only resolves the token it observed when it started. A signal244 * listing only resolves the token it observed when it started. A signal
245 * that arrives mid-listing therefore survives it and re-lists. */245 * that arrives mid-listing therefore survives it and re-lists.
246 *
247 * NOT from inside `ctx.transaction(...)`, though: the write takes a
248 * non-reentrant executor permit, and on sqlite an open BEGIN blocking on
249 * that permit can silently drop a concurrent catalog rebuild. Call it
250 * before the transaction, or after it commits. */
246 readonly markToolsStale: (ref: ConnectionRef) => Effect.Effect<void, StorageFailure>;251 readonly markToolsStale: (ref: ConnectionRef) => Effect.Effect<void, StorageFailure>;
247 /** Resolve a connection's value through its provider (and OAuth refresh).252 /** Resolve a connection's value through its provider (and OAuth refresh).
248 * null if the provider can't produce one. */253 * null if the provider can't produce one. */
packages/core/sdk/src/tools-sync-scope.test.ts+2−1
@@ -310,7 +310,8 @@ const makeHarness = () => · 309 unmodified lines
310 ? Effect.die(`no connection row named ${name}`)310 ? Effect.die(`no connection row named ${name}`)
311 : Effect.succeed({311 : Effect.succeed({
312 syncedAt: row.tools_synced_at == null ? null : Number(row.tools_synced_at),312 syncedAt: row.tools_synced_at == null ? null : Number(row.tools_synced_at),
313 staleToken: row.tools_stale_token ?? null,313 staleToken:
314 row.tools_stale_token == null ? null : String(row.tools_stale_token),
314 failures: row.tools_sync_failures == null ? 0 : Number(row.tools_sync_failures),315 failures: row.tools_sync_failures == null ? 0 : Number(row.tools_sync_failures),
315 retryAt:316 retryAt:
316 row.tools_sync_retry_at == null ? null : Number(row.tools_sync_retry_at),317 row.tools_sync_retry_at == null ? null : Number(row.tools_sync_retry_at),