| author |  Theo Browne <me@t3.gg> 2026-07-17 14:26:22-07:00 |
| committer |  GitHub <noreply@github.com> 2026-07-17 21:26:22+00:00 |
| log | 31ca9e553112ba336bfb3f6c04951f1086f37b0c |
| tree | d3858e2804ee1e6d8749412ad2ebdbb29898471d |
| parent | 73bf127d71fbe7fa226cdf021a8a14cf272b2d8e |
| signature | Signed by PGP key B5690EEEBB952194 |
fix(server): skip undecodable provider runtime rows when listing sessions (#3951)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Julius Marminge <julius0216@outlook.com>
Co-authored-by: Julius Marminge <jmarminge@gmail.com>
Co-authored-by: codex <codex@users.noreply.github.com>2 files changed, 38 insertions(+), 16 deletions(-)
apps
server
apps/server/src/persistence/ProviderSessionRuntime.ts+20-6
| ... | ... | @@ -1,7 +1,9 @@ |
| 1 | import * as Arr from "effect/Array"; | |
| 1 | 2 | import * as Context from "effect/Context"; |
| 2 | 3 | import * as Effect from "effect/Effect"; |
| 3 | 4 | import * as Layer from "effect/Layer"; |
| 4 | 5 | import * as Option from "effect/Option"; |
| 6 | import * as Result from "effect/Result"; | |
| 5 | 7 | import * as Schema from "effect/Schema"; |
| 6 | 8 | import * as Struct from "effect/Struct"; |
| 7 | 9 | import * as SqlClient from "effect/unstable/sql/SqlClient"; |
| ... | ... | @@ -280,18 +282,30 @@ export const make = Effect.gen(function* () { |
| 280 | 282 | ), |
| 281 | 283 | ), |
| 282 | 284 | Effect.flatMap((rows) => |
| 285 | // Skip rows that no longer decode (e.g. written by an older build) | |
| 286 | // instead of failing the whole list — one stale row must not disable | |
| 287 | // every consumer that enumerates sessions, such as the reaper. | |
| 283 | 288 | Effect.forEach(rows, (row) => |
| 284 | 289 | decodeRuntimeRow(row).pipe( |
| 285 | Effect.mapError((cause) => | |
| 286 | PersistenceDecodeError.fromSchemaError( | |
| 287 | "ProviderSessionRuntimeRepository.list:decodeRows", | |
| 288 | cause, | |
| 289 | { threadId: row.threadId }, | |
| 290 | ), | |
| 290 | Effect.map(Option.some), | |
| 291 | Effect.catch((cause) => | |
| 292 | Effect.logWarning("provider.session.runtime.row-skipped", { | |
| 293 | threadId: row.threadId, | |
| 294 | error: PersistenceDecodeError.fromSchemaError( | |
| 295 | "ProviderSessionRuntimeRepository.list:decodeRows", | |
| 296 | cause, | |
| 297 | { threadId: row.threadId }, | |
| 298 | ).message, | |
| 299 | }).pipe(Effect.as(Option.none<ProviderSessionRuntime>())), | |
| 291 | 300 | ), |
| 292 | 301 | ), |
| 293 | 302 | ), |
| 294 | 303 | ), |
| 304 | Effect.map((decoded) => | |
| 305 | Arr.filterMap(decoded, (row) => | |
| 306 | Option.isSome(row) ? Result.succeed(row.value) : Result.failVoid, | |
| 307 | ), | |
| 308 | ), | |
| 295 | 309 | ); |
| 296 | 310 | |
| 297 | 311 | const deleteByThreadId: ProviderSessionRuntimeRepository["Service"]["deleteByThreadId"] = ( |
apps/server/src/persistence/RepositoryErrorCorrelation.test.ts+18-10
| ... | ... | @@ -182,7 +182,7 @@ describe("persistence error correlation", () => { |
| 182 | 182 | }).pipe(Effect.provide(authPairingLinkLayer)), |
| 183 | 183 | ); |
| 184 | 184 | |
| 185 | it.effect("correlates provider runtime SQL and per-row decode failures by thread", () => | |
| 185 | it.effect("skips undecodable provider runtime rows and correlates SQL failures by thread", () => | |
| 186 | 186 | Effect.gen(function* () { |
| 187 | 187 | const runtimes = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; |
| 188 | 188 | const sql = yield* SqlClient.SqlClient; |
| ... | ... | @@ -215,16 +215,24 @@ describe("persistence error correlation", () => { |
| 215 | 215 | ) |
| 216 | 216 | `; |
| 217 | 217 | |
| 218 | const decodeError = yield* Effect.flip(runtimes.list()); | |
| 219 | assert.instanceOf(decodeError, PersistenceErrors.PersistenceDecodeError); | |
| 220 | assert.deepStrictEqual(decodeError.correlation, { threadId }); | |
| 221 | assert.equal( | |
| 222 | decodeError.message, | |
| 223 | `Decode error in ProviderSessionRuntimeRepository.list:decodeRows: ${decodeError.issue}`, | |
| 218 | const validThreadId = ThreadId.make("thread-valid"); | |
| 219 | yield* runtimes.upsert({ | |
| 220 | threadId: validThreadId, | |
| 221 | providerName: "codex", | |
| 222 | providerInstanceId: null, | |
| 223 | adapterKey: "codex", | |
| 224 | runtimeMode: "full-access", | |
| 225 | status: "running", | |
| 226 | lastSeenAt, | |
| 227 | resumeCursor: null, | |
| 228 | runtimePayload: null, | |
| 229 | }); | |
| 230 | ||
| 231 | const listed = yield* runtimes.list(); | |
| 232 | assert.deepStrictEqual( | |
| 233 | listed.map((runtime) => runtime.threadId), | |
| 234 | [validThreadId], | |
| 224 | 235 | ); |
| 225 | assert.notInclude(decodeError.issue, runtimePayload); | |
| 226 | assert.notInclude(decodeError.message, runtimePayload); | |
| 227 | assert.notInclude(decodeError.message, lastSeenAt); | |
| 228 | 236 | |
| 229 | 237 | yield* sql`DROP TABLE provider_session_runtime`; |
| 230 | 238 | const sqlFailure = yield* Effect.flip( |