| author | |
| committer | |
| log | 1c33f3b2b2976b7f4e26698ed69c549de687a97e |
| tree | 9421093df85dff75a96651018e73d2a1035dd894 |
| parent | 31f9d832f4f019ff565beafbd6bb18451fa7039d |
| signature | Signed by PGP key B5690EEEBB952194 |
2 files changed, 140 insertions(+), 27 deletions(-)
apps/mobile/src/connection/environment-cache-store.test.ts+93-4| ... | @@ -388,16 +388,105 @@ describe("cooperative shell cache encoding", () => { | ... | @@ -388,16 +388,105 @@ describe("cooperative shell cache encoding", () => { |
| 388 | }), | 388 | }), |
| 389 | ); | 389 | ); |
| 390 | 390 | ||
| 391 | it.effect("still rejects invalid rows after an encoding batch yields", () => | 391 | it.effect("preserves optional fields and dates across chunk boundaries and archived rows", () => |
| 392 | Effect.gen(function* () { | 392 | Effect.gen(function* () { |
| 393 | const invalid = { | 393 | const base = SHELL_SNAPSHOT.threads[0]!; |
| 394 | const { titleRegeneration: _, unsettledAt: __, ...withoutOptional } = base; | ||
| 395 | const row = (index: number) => | ||
| 396 | index % 3 === 0 | ||
| 397 | ? { ...withoutOptional, id: ThreadId.make(`row-${index}`) } | ||
| 398 | : { | ||
| 399 | ...base, | ||
| 400 | id: ThreadId.make(`row-${index}`), | ||
| 401 | snoozedUntil: DateTime.add(NOW, { minutes: index }), | ||
| 402 | pinnedAt: index % 2 === 0 ? null : NOW, | ||
| 403 | }; | ||
| 404 | const mixed = { | ||
| 394 | ...stored, | 405 | ...stored, |
| 395 | snapshot: { | 406 | snapshot: { |
| 396 | ...stored.snapshot, | 407 | ...stored.snapshot, |
| 397 | threads: [...stored.snapshot.threads, { ...stored.snapshot.threads[0]!, itemCount: -1 }], | 408 | threads: Array.from({ length: 65 }, (_, index) => row(index)), |
| 409 | archivedThreads: Array.from({ length: 33 }, (_, index) => ({ | ||
| 410 | ...row(index + 100), | ||
| 411 | archivedAt: DateTime.add(NOW, { hours: index }), | ||
| 412 | })), | ||
| 398 | }, | 413 | }, |
| 399 | }; | 414 | }; |
| 400 | expect(yield* Effect.isFailure(encodeStoredShellSnapshot(invalid))).toBe(true); | 415 | const actual = yield* encodeStoredShellSnapshot(mixed); |
| 416 | expect(actual).toBe(yield* encodeOriginal(mixed)); | ||
| 417 | const parsed = JSON.parse(actual); | ||
| 418 | expect(parsed.snapshot.threads).toHaveLength(65); | ||
| 419 | expect(parsed.snapshot.threads[63]).not.toHaveProperty("titleRegeneration"); | ||
| 420 | expect(parsed.snapshot.threads[63]).not.toHaveProperty("unsettledAt"); | ||
| 421 | expect(parsed.snapshot.threads[64].snoozedUntil).toBe( | ||
| 422 | DateTime.formatIso(DateTime.add(NOW, { minutes: 64 })), | ||
| 423 | ); | ||
| 424 | expect(parsed.snapshot.archivedThreads).toHaveLength(33); | ||
| 425 | expect(parsed.snapshot.archivedThreads[32].archivedAt).toBe( | ||
| 426 | DateTime.formatIso(DateTime.add(NOW, { hours: 32 })), | ||
| 427 | ); | ||
| 428 | }), | ||
| 429 | ); | ||
| 430 | |||
| 431 | it.effect("yields to a host timer between every 32-row chunk", () => | ||
| 432 | Effect.gen(function* () { | ||
| 433 | const setTimer = vi.spyOn(globalThis, "setTimeout"); | ||
| 434 | try { | ||
| 435 | yield* encodeStoredShellSnapshot({ | ||
| 436 | ...stored, | ||
| 437 | snapshot: { | ||
| 438 | ...stored.snapshot, | ||
| 439 | threads: Array.from({ length: 65 }, () => stored.snapshot.threads[0]!), | ||
| 440 | archivedThreads: Array.from({ length: 33 }, () => stored.snapshot.archivedThreads[0]!), | ||
| 441 | }, | ||
| 442 | }); | ||
| 443 | // Five row chunks (3 active + 2 archived): four yields between them, one before the envelope. | ||
| 444 | expect(setTimer).toHaveBeenCalledTimes(5); | ||
| 445 | } finally { | ||
| 446 | vi.restoreAllMocks(); | ||
| 447 | } | ||
| 448 | }), | ||
| 449 | ); | ||
| 450 | |||
| 451 | it.effect("still rejects invalid active and archived rows after an encoding chunk yields", () => | ||
| 452 | Effect.gen(function* () { | ||
| 453 | const invalidRow = { ...stored.snapshot.threads[0]!, itemCount: -1 }; | ||
| 454 | const invalidActive = { | ||
| 455 | ...stored, | ||
| 456 | snapshot: { ...stored.snapshot, threads: [...stored.snapshot.threads, invalidRow] }, | ||
| 457 | }; | ||
| 458 | const invalidArchived = { | ||
| 459 | ...stored, | ||
| 460 | snapshot: { ...stored.snapshot, archivedThreads: [invalidRow] }, | ||
| 461 | }; | ||
| 462 | const invalidEnvelope = { ...stored, snapshot: { ...stored.snapshot, snapshotSequence: -1 } }; | ||
| 463 | expect(yield* Effect.isFailure(encodeStoredShellSnapshot(invalidActive))).toBe(true); | ||
| 464 | expect(yield* Effect.isFailure(encodeStoredShellSnapshot(invalidArchived))).toBe(true); | ||
| 465 | expect(yield* Effect.isFailure(encodeOriginal(invalidEnvelope))).toBe(true); | ||
| 466 | expect(yield* Effect.isFailure(encodeStoredShellSnapshot(invalidEnvelope))).toBe(true); | ||
| 467 | }), | ||
| 468 | ); | ||
| 469 | |||
| 470 | it.effect("keeps concurrent saves independent", () => | ||
| 471 | Effect.gen(function* () { | ||
| 472 | const other = { | ||
| 473 | ...stored, | ||
| 474 | environmentId: EnvironmentId.make("environment-2"), | ||
| 475 | snapshot: { | ||
| 476 | ...stored.snapshot, | ||
| 477 | threads: stored.snapshot.threads.slice(0, 40).map((thread) => ({ | ||
| 478 | ...thread, | ||
| 479 | title: `Other ${thread.id}`, | ||
| 480 | })), | ||
| 481 | archivedThreads: [], | ||
| 482 | }, | ||
| 483 | }; | ||
| 484 | const [first, second] = yield* Effect.all( | ||
| 485 | [encodeStoredShellSnapshot(stored), encodeStoredShellSnapshot(other)], | ||
| 486 | { concurrency: "unbounded" }, | ||
| 487 | ); | ||
| 488 | expect(first).toBe(yield* encodeOriginal(stored)); | ||
| 489 | expect(second).toBe(yield* encodeOriginal(other)); | ||
| 401 | }), | 490 | }), |
| 402 | ); | 491 | ); |
| 403 | 492 |
apps/mobile/src/connection/shell-cache-encoding.ts+47-23| ... | @@ -3,36 +3,60 @@ import { OrchestrationV2ThreadShellJson } from "@t3tools/contracts"; | ... | @@ -3,36 +3,60 @@ import { OrchestrationV2ThreadShellJson } from "@t3tools/contracts"; |
| 3 | import * as Effect from "effect/Effect"; | 3 | import * as Effect from "effect/Effect"; |
| 4 | import * as Schema from "effect/Schema"; | 4 | import * as Schema from "effect/Schema"; |
| 5 | 5 | ||
| 6 | const ROWS_PER_CHUNK = 32; | ||
| 7 | |||
| 6 | // Effect's cooperative yield resumes through microtasks on React Native. A host | 8 | // Effect's cooperative yield resumes through microtasks on React Native. A host |
| 7 | // timer lets input and rendering run between batches of shell rows instead. | 9 | // timer lets input and rendering run between chunks of shell rows instead. |
| 8 | const yieldToHost = Effect.callback<void>((resume) => { | 10 | const yieldToHost = Effect.callback<void>((resume) => { |
| 9 | const timer = setTimeout(() => resume(Effect.void), 0); | 11 | const timer = setTimeout(() => resume(Effect.void), 0); |
| 10 | return Effect.sync(() => clearTimeout(timer)); | 12 | return Effect.sync(() => clearTimeout(timer)); |
| 11 | }); | 13 | }); |
| 12 | 14 | ||
| 13 | // Share the row budget across saves and environments; this bounds batches, not elapsed time. | 15 | // Rows go through the canonical row codec in chunks. The envelope codec then |
| 14 | let encodedRows = 0; | 16 | // validates and encodes everything else, passing the encoded rows through as-is. |
| 15 | const CooperativeThreadShell = OrchestrationV2ThreadShellJson.pipe( | 17 | const encodeThreadChunk = Schema.encodeEffect(Schema.Array(OrchestrationV2ThreadShellJson)); |
| 16 | Schema.middlewareEncoding<typeof OrchestrationV2ThreadShellJson, never>((effect) => | 18 | const EncodedRows = Schema.Array(Schema.Unknown); |
| 17 | Effect.flatMap(effect, (value) => { | 19 | const encodeEnvelope = Schema.encodeEffect( |
| 18 | encodedRows++; | 20 | Schema.fromJsonString( |
| 19 | if (encodedRows < 32) return Effect.succeed(value); | 21 | StoredOrchestrationShellSnapshot.mapFields((fields) => ({ |
| 20 | encodedRows = 0; | 22 | ...fields, |
| 21 | return Effect.as(yieldToHost, value); | 23 | snapshot: fields.snapshot.mapFields((snapshotFields) => ({ |
| 22 | }), | 24 | ...snapshotFields, |
| 25 | threads: EncodedRows, | ||
| 26 | archivedThreads: EncodedRows, | ||
| 27 | })), | ||
| 28 | })), | ||
| 23 | ), | 29 | ), |
| 24 | ); | 30 | ); |
| 25 | 31 | ||
| 26 | const CooperativeStoredShellSnapshot = StoredOrchestrationShellSnapshot.mapFields((fields) => ({ | 32 | /** |
| 27 | ...fields, | 33 | * Encode the shell cache payload exactly like `fromJsonString(StoredOrchestrationShellSnapshot)`, |
| 28 | snapshot: fields.snapshot.mapFields((snapshotFields) => ({ | 34 | * yielding to the host between bounded chunks of thread rows. The final envelope encode and |
| 29 | ...snapshotFields, | 35 | * JSON stringify still run synchronously over the whole payload. |
| 30 | threads: Schema.Array(CooperativeThreadShell), | 36 | */ |
| 31 | archivedThreads: Schema.Array(CooperativeThreadShell), | 37 | export const encodeStoredShellSnapshot = Effect.fnUntraced(function* ( |
| 32 | })), | 38 | stored: typeof StoredOrchestrationShellSnapshot.Type, |
| 33 | })); | 39 | ) { |
| 40 | let hasWorked = false; | ||
| 41 | const yieldBetweenChunks = Effect.suspend(() => { | ||
| 42 | if (hasWorked) return yieldToHost; | ||
| 43 | hasWorked = true; | ||
| 44 | return Effect.void; | ||
| 45 | }); | ||
| 46 | const encodeRows = (rows: ReadonlyArray<OrchestrationV2ThreadShellJson>) => | ||
| 47 | Effect.gen(function* () { | ||
| 48 | const encoded: Array<unknown> = []; | ||
| 49 | for (let start = 0; start < rows.length; start += ROWS_PER_CHUNK) { | ||
| 50 | yield* yieldBetweenChunks; | ||
| 51 | const chunk = yield* encodeThreadChunk(rows.slice(start, start + ROWS_PER_CHUNK)); | ||
| 52 | for (const row of chunk) encoded.push(row); | ||
| 53 | } | ||
| 54 | return encoded; | ||
| 55 | }); | ||
| 34 | 56 | ||
| 35 | /** Preserve the cache codec's validation and transforms while yielding during large saves. */ | 57 | const { snapshot } = stored; |
| 36 | export const encodeStoredShellSnapshot = Schema.encodeEffect( | 58 | const threads = yield* encodeRows(snapshot.threads); |
| 37 | Schema.fromJsonString(CooperativeStoredShellSnapshot), | 59 | const archivedThreads = yield* encodeRows(snapshot.archivedThreads); |
| 38 | ); | 60 | yield* yieldBetweenChunks; |
| 61 | return yield* encodeEnvelope({ ...stored, snapshot: { ...snapshot, threads, archivedThreads } }); | ||
| 62 | }); |