Skip to content
2 changes: 1 addition & 1 deletion apps/mobile/src/features/threads/ThreadDetailScreen.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
106 changes: 106 additions & 0 deletions apps/server/src/persistence/DatabaseStorage.test.ts
Original file line number Diff line number Diff line change
@@ -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";
Expand Down Expand Up @@ -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<number>();
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.pipe(Layer.provideMerge(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<number>();
const finish = yield* Queue.unbounded<void>();
const writeFinished = yield* Deferred.make<void>();
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.pipe(Layer.provideMerge(NodeServices.layer)))),
);
40 changes: 32 additions & 8 deletions apps/server/src/persistence/DatabaseStorage.ts
Original file line number Diff line number Diff line change
@@ -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";
Expand All @@ -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,
Expand All @@ -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<DatabaseStorageStatus | null>(null);

Expand Down Expand Up @@ -89,21 +94,40 @@ const make = Effect.gen(function* () {
Effect.catch((cause) => Effect.logWarning("Database space reclamation failed", { cause })),
);

const compact = maintenance.compactEventStore.pipe(
Effect.tap((result) =>
result.deletedEventCount > 0 || result.deletedReceiptCount > 0
? Effect.logInfo("Compacted orchestration history", result)
: Effect.void,
),
// 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 = 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 }),
),
);
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)),
),
),
),
);
Expand Down
Loading