From 0094f6ae666a30676c671c64ccb4cc5b089102fe Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 18:47:15 +0000 Subject: [PATCH 1/5] fix(server): keep compaction cadence across restarts Rebuild the cadence portion of PR #21192 by Alyx-ML; contributor credit retained in the replacement PR. --- .../server/src/persistence/DatabaseStorage.ts | 33 +++++++++++++++++-- 1 file changed, 31 insertions(+), 2 deletions(-) diff --git a/apps/server/src/persistence/DatabaseStorage.ts b/apps/server/src/persistence/DatabaseStorage.ts index 3866020c1c1c..56924a2c004b 100644 --- a/apps/server/src/persistence/DatabaseStorage.ts +++ b/apps/server/src/persistence/DatabaseStorage.ts @@ -1,7 +1,10 @@ import type { DatabaseStorageStatus } from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Equal from "effect/Equal"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Schedule from "effect/Schedule"; @@ -16,6 +19,7 @@ import { forkParked } from "../serverActivation.ts"; /** Pages returned per reclaim pass; one minute of churn, not a backlog. */ const RECLAIM_PAGES_PER_PASS = 1024; +const COMPACTION_INTERVAL = Duration.hours(6); export class DatabaseStorage extends Context.Service< DatabaseStorage, @@ -37,6 +41,7 @@ const make = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; const statusRef = yield* SubscriptionRef.make(null); @@ -89,21 +94,45 @@ const make = Effect.gen(function* () { Effect.catch((cause) => Effect.logWarning("Database space reclamation failed", { cause })), ); + // A pass reads the whole event store, gigabytes on a long-lived install. + // Remembering when one last finished keeps the six-hour cadence across + // restarts instead of rescanning a minute into every launch. + const compactedAtPath = path.join(path.dirname(config.dbPath), "compacted-at"); const compact = maintenance.compactEventStore.pipe( Effect.tap((result) => result.deletedEventCount > 0 || result.deletedReceiptCount > 0 ? Effect.logInfo("Compacted orchestration history", result) : Effect.void, ), + Effect.tap(() => + Clock.currentTimeMillis.pipe( + Effect.flatMap((now) => fs.writeFileString(compactedAtPath, String(now))), + Effect.ignore, + ), + ), Effect.catch((cause) => Effect.logWarning("Orchestration history compaction failed", { cause }), ), ); + const lastCompactedAt = yield* fs.readFileString(compactedAtPath).pipe( + Effect.map(Number), + Effect.orElseSucceed(() => Number.NaN), + ); + const intervalMs = Duration.toMillis(COMPACTION_INTERVAL); + const untilDueMs = Number.isFinite(lastCompactedAt) + ? lastCompactedAt + intervalMs - (yield* Clock.currentTimeMillis) + : 0; + const firstCompactionDelay = Duration.millis( + Math.min(intervalMs, Math.max(Duration.toMillis(Duration.minutes(1)), untilDueMs)), + ); yield* forkParked( - Effect.sleep("1 minute").pipe( + Effect.sleep(firstCompactionDelay).pipe( Effect.andThen( - compact.pipe(Effect.andThen(publishStatus), Effect.repeat(Schedule.spaced("6 hours"))), + compact.pipe( + Effect.andThen(publishStatus), + Effect.repeat(Schedule.spaced(COMPACTION_INTERVAL)), + ), ), ), ); From 2858a51cf90ed93e47b9fb3eaed35202f8ced2e2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 18:52:35 +0000 Subject: [PATCH 2/5] test(server): cover persisted compaction cadence --- .../src/persistence/DatabaseStorage.test.ts | 106 ++++++++++++++++++ 1 file changed, 106 insertions(+) diff --git a/apps/server/src/persistence/DatabaseStorage.test.ts b/apps/server/src/persistence/DatabaseStorage.test.ts index 7196f18f40ed..84278fc54bd7 100644 --- a/apps/server/src/persistence/DatabaseStorage.test.ts +++ b/apps/server/src/persistence/DatabaseStorage.test.ts @@ -1,10 +1,13 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; +import * as Clock from "effect/Clock"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import * as SqlClient from "effect/sql/SqlClient"; @@ -108,3 +111,106 @@ it.effect("publishes nothing when doctor would repair a different database", () ), ), ); + +const compactionLayer = ( + compactEventStore: ProjectionMaintenance.ProjectionMaintenanceV2["Service"]["compactEventStore"], +) => + DatabaseStorage.layer.pipe( + Layer.provide( + Layer.succeed(ProjectionMaintenance.ProjectionMaintenanceV2, { + compactEventStore, + verify: Effect.die("unexpected verification"), + rebuild: Effect.die("unexpected rebuild"), + }), + ), + Layer.provide(Sqlite.layerMemory), + ); + +const minute = 60_000; +const hour = 60 * minute; +const startupTime = Date.UTC(2026, 9, 19); +const emptyCompaction = { deletedEventCount: 0, deletedReceiptCount: 0, reclaimableBytes: 0 }; + +it.effect.each([ + { name: "recent success", saved: String(startupTime - 2 * hour), delay: 4 * hour }, + { name: "overdue success", saved: String(startupTime - 7 * hour), delay: minute }, + { name: "nearly due success", saved: String(startupTime - 6 * hour + 1), delay: minute }, + { name: "future timestamp", saved: String(startupTime + hour), delay: 6 * hour }, + { name: "missing timestamp", saved: undefined, delay: minute }, + { name: "invalid timestamp", saved: "not a timestamp", delay: minute }, + { name: "unreadable timestamp", saved: undefined, delay: minute }, +])("schedules startup compaction after $name", ({ name, saved, delay }) => + Effect.gen(function* () { + yield* TestClock.setTime(startupTime); + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const config = yield* ServerConfig.ServerConfig; + const timestampPath = path.join(path.dirname(config.dbPath), "compacted-at"); + if (name === "unreadable timestamp") yield* fs.makeDirectory(timestampPath); + else if (saved !== undefined) yield* fs.writeFileString(timestampPath, saved); + const started = yield* Deferred.make(); + const compact = Clock.currentTimeMillis.pipe( + Effect.flatMap((now) => Deferred.succeed(started, now)), + Effect.as(emptyCompaction), + ); + yield* Effect.gen(function* () { + yield* TestClock.adjust(delay - 1); + assert.isFalse(yield* Deferred.isDone(started)); + yield* TestClock.adjust(1); + assert.equal(yield* Deferred.await(started), startupTime + delay); + }).pipe(Effect.provide(compactionLayer(compact))); + }).pipe(Effect.provide(testConfig), Effect.provide(NodeServices.layer)), +); + +it.effect.each(["success", "compaction failure", "timestamp write failure"] as const)( + "keeps the six-hour cadence after %s and remembers only successful completion", + (outcome) => + Effect.gen(function* () { + yield* TestClock.setTime(startupTime); + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const config = yield* ServerConfig.ServerConfig; + const timestampPath = path.join(path.dirname(config.dbPath), "compacted-at"); + const previous = String(startupTime - 7 * hour); + if (outcome === "timestamp write failure") yield* fs.makeDirectory(timestampPath); + else yield* fs.writeFileString(timestampPath, previous); + const starts = yield* Queue.unbounded(); + const finish = yield* Queue.unbounded(); + const writeFinished = yield* Deferred.make(); + const compact = Effect.gen(function* () { + yield* Queue.offer(starts, yield* Clock.currentTimeMillis); + yield* Queue.take(finish); + if (outcome === "compaction failure") { + return yield* new ProjectionMaintenance.ProjectionMaintenanceError({ operation: "test" }); + } + return emptyCompaction; + }); + yield* Effect.gen(function* () { + yield* TestClock.adjust(minute); + assert.equal(yield* Queue.take(starts), startupTime + minute); + yield* TestClock.adjust(2 * minute); + if (outcome !== "timestamp write failure") { + assert.equal(yield* fs.readFileString(timestampPath), previous); + } + yield* Queue.offer(finish, undefined); + if (outcome !== "compaction failure") yield* Deferred.await(writeFinished); + yield* TestClock.adjust(6 * hour - 1); + assert.equal(yield* Queue.size(starts), 0); + yield* TestClock.adjust(1); + assert.equal(yield* Queue.take(starts), startupTime + 3 * minute + 6 * hour); + }).pipe( + Effect.provide(compactionLayer(compact)), + Effect.provideService(FileSystem.FileSystem, { + ...fs, + writeFileString: (file, content, options) => + fs + .writeFileString(file, content, options) + .pipe(Effect.ensuring(Deferred.succeed(writeFinished, undefined))), + }), + ); + if (outcome !== "timestamp write failure") { + const saved = Number(yield* fs.readFileString(timestampPath)); + assert.equal(saved, outcome === "success" ? startupTime + 3 * minute : Number(previous)); + } + }).pipe(Effect.provide(testConfig), Effect.provide(NodeServices.layer)), +); From cc25e29b0565a1af5bb2b3a2a9b5ddcc1518c3d4 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 18:58:53 +0000 Subject: [PATCH 3/5] refactor(server): simplify compaction completion persistence --- .../server/src/persistence/DatabaseStorage.ts | 25 ++++++++----------- 1 file changed, 10 insertions(+), 15 deletions(-) diff --git a/apps/server/src/persistence/DatabaseStorage.ts b/apps/server/src/persistence/DatabaseStorage.ts index 56924a2c004b..2dfcd22c0c85 100644 --- a/apps/server/src/persistence/DatabaseStorage.ts +++ b/apps/server/src/persistence/DatabaseStorage.ts @@ -94,22 +94,17 @@ const make = Effect.gen(function* () { Effect.catch((cause) => Effect.logWarning("Database space reclamation failed", { cause })), ); - // A pass reads the whole event store, gigabytes on a long-lived install. - // Remembering when one last finished keeps the six-hour cadence across - // restarts instead of rescanning a minute into every launch. + // Even a pass that deletes nothing scans the event store. Persist completion + // so restarting does not trigger another scan before the six-hour interval. const compactedAtPath = path.join(path.dirname(config.dbPath), "compacted-at"); - const compact = maintenance.compactEventStore.pipe( - Effect.tap((result) => - result.deletedEventCount > 0 || result.deletedReceiptCount > 0 - ? Effect.logInfo("Compacted orchestration history", result) - : Effect.void, - ), - Effect.tap(() => - Clock.currentTimeMillis.pipe( - Effect.flatMap((now) => fs.writeFileString(compactedAtPath, String(now))), - Effect.ignore, - ), - ), + const compact = Effect.gen(function* () { + const result = yield* maintenance.compactEventStore; + const completedAt = yield* Clock.currentTimeMillis; + yield* fs.writeFileString(compactedAtPath, String(completedAt)).pipe(Effect.ignore); + if (result.deletedEventCount > 0 || result.deletedReceiptCount > 0) { + yield* Effect.logInfo("Compacted orchestration history", result); + } + }).pipe( Effect.catch((cause) => Effect.logWarning("Orchestration history compaction failed", { cause }), ), From 88465a1d8935ec501cbc04bd680ee98e993d4eb5 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 19 Oct 2026 19:00:38 +0000 Subject: [PATCH 4/5] test(server): provide cadence fixtures through one layer --- apps/server/src/persistence/DatabaseStorage.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/server/src/persistence/DatabaseStorage.test.ts b/apps/server/src/persistence/DatabaseStorage.test.ts index 84278fc54bd7..23dda5aabc57 100644 --- a/apps/server/src/persistence/DatabaseStorage.test.ts +++ b/apps/server/src/persistence/DatabaseStorage.test.ts @@ -159,7 +159,7 @@ it.effect.each([ yield* TestClock.adjust(1); assert.equal(yield* Deferred.await(started), startupTime + delay); }).pipe(Effect.provide(compactionLayer(compact))); - }).pipe(Effect.provide(testConfig), Effect.provide(NodeServices.layer)), + }).pipe(Effect.provide(testConfig.pipe(Layer.provideMerge(NodeServices.layer)))), ); it.effect.each(["success", "compaction failure", "timestamp write failure"] as const)( @@ -212,5 +212,5 @@ it.effect.each(["success", "compaction failure", "timestamp write failure"] as c const saved = Number(yield* fs.readFileString(timestampPath)); assert.equal(saved, outcome === "success" ? startupTime + 3 * minute : Number(previous)); } - }).pipe(Effect.provide(testConfig), Effect.provide(NodeServices.layer)), + }).pipe(Effect.provide(testConfig.pipe(Layer.provideMerge(NodeServices.layer)))), ); From 4c62cee3a93f13ca54c7b7c297f0a57b060097bd Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 20 Oct 2026 03:05:47 +0000 Subject: [PATCH 5/5] fix(mobile): import thread screen callback hook --- apps/mobile/src/features/threads/ThreadDetailScreen.tsx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 6d354e57c508..ac22a8b7d9e1 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -58,7 +58,7 @@ import type { QueuedRunEdit } from "../../state/queued-run-edit"; import type { FollowUpBehavior } from "../../lib/followUpBehavior"; import type { ActiveTurnComposerAction } from "@t3tools/client-runtime/state/composer-dispatch"; import * as Haptics from "expo-haptics"; -import { useContext, useEffect, useLayoutEffect, useRef, useState } from "react"; +import { useCallback, useContext, useEffect, useLayoutEffect, useRef, useState } from "react"; import { AppState, Keyboard,