From 8efbb31725f8500e1aff76655b04a3fab7b064fe Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 26 Feb 2026 10:35:37 -0800 Subject: [PATCH] use effect/Cache instead of ad-hoc map caches --- apps/server/src/git/Layers/GitCore.ts | 121 ++++++++++-------- .../Layers/ProviderCommandReactor.ts | 53 +++----- .../Layers/ProviderRuntimeIngestion.ts | 108 ++++++++++------ .../src/provider/Layers/ProviderService.ts | 24 +--- 4 files changed, 155 insertions(+), 151 deletions(-) diff --git a/apps/server/src/git/Layers/GitCore.ts b/apps/server/src/git/Layers/GitCore.ts index 0624bd91fb1c9a1a39b2bcfeb6ac29101c4116ae..fdbf3c8bc30b948eacc861b1c140f93eb469c6dd 100644 --- a/apps/server/src/git/Layers/GitCore.ts +++ b/apps/server/src/git/Layers/GitCore.ts @@ -1,11 +1,19 @@ -import { Effect, FileSystem, Layer, Path } from "effect"; +import { Cache, Data, Duration, Effect, Exit, FileSystem, Layer, Path } from "effect"; import { GitCommandError } from "../Errors.ts"; import { GitService } from "../Services/GitService.ts"; import { GitCore, type GitCoreShape } from "../Services/GitCore.ts"; -const STATUS_UPSTREAM_REFRESH_INTERVAL_MS = 15_000; -const STATUS_UPSTREAM_REFRESH_TIMEOUT_MS = 5_000; +const STATUS_UPSTREAM_REFRESH_INTERVAL = Duration.seconds(15); +const STATUS_UPSTREAM_REFRESH_TIMEOUT = Duration.seconds(5); +const STATUS_UPSTREAM_REFRESH_CACHE_CAPACITY = 2_048; + +class StatusUpstreamRefreshCacheKey extends Data.Class<{ + cwd: string; + upstreamRef: string; + remoteName: string; + upstreamBranch: string; +}> {} interface ExecuteGitOptions { timeoutMs?: number | undefined; @@ -13,10 +21,6 @@ interface ExecuteGitOptions { fallbackErrorMessage?: string | undefined; } -function trimStdout(value: string): string { - return value.trim(); -} - function parseBranchAb(value: string): { ahead: number; behind: number } { const match = value.match(/^\+(\d+)\s+-(\d+)$/); if (!match) return { ahead: 0, behind: 0 }; @@ -96,7 +100,6 @@ const makeGitCore = Effect.gen(function* () { const git = yield* GitService; const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; - const lastUpstreamRefreshAt = new Map(); const executeGit = ( operation: string, @@ -162,14 +165,13 @@ const makeGitCore = Effect.gen(function* () { GitCommandError > => Effect.gen(function* () { - const upstreamRef = trimStdout( - yield* runGitStdout( - "GitCore.resolveCurrentUpstream", - cwd, - ["rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{upstream}"], - true, - ), - ); + const upstreamRef = yield* runGitStdout( + "GitCore.resolveCurrentUpstream", + cwd, + ["rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{upstream}"], + true, + ).pipe(Effect.map((stdout) => stdout.trim())); + if (upstreamRef.length === 0 || upstreamRef === "@{upstream}") { return null; } @@ -215,36 +217,45 @@ const makeGitCore = Effect.gen(function* () { ["fetch", "--quiet", "--no-tags", upstream.remoteName, refspec], { allowNonZeroExit: true, - timeoutMs: STATUS_UPSTREAM_REFRESH_TIMEOUT_MS, + timeoutMs: Duration.toMillis(STATUS_UPSTREAM_REFRESH_TIMEOUT), }, ).pipe(Effect.asVoid); }; + const statusUpstreamRefreshCache = yield* Cache.makeWith({ + capacity: STATUS_UPSTREAM_REFRESH_CACHE_CAPACITY, + lookup: (cacheKey: StatusUpstreamRefreshCacheKey) => + Effect.gen(function* () { + yield* fetchUpstreamRefForStatus(cacheKey.cwd, { + upstreamRef: cacheKey.upstreamRef, + remoteName: cacheKey.remoteName, + upstreamBranch: cacheKey.upstreamBranch, + }); + return true as const; + }), + // Keep successful refreshes warm; drop failures immediately so next request can retry. + timeToLive: (exit) => (Exit.isSuccess(exit) ? STATUS_UPSTREAM_REFRESH_INTERVAL : Duration.zero), + }); + const refreshStatusUpstreamIfStale = (cwd: string): Effect.Effect => Effect.gen(function* () { const upstream = yield* resolveCurrentUpstream(cwd); - if (!upstream) { - return; - } - - const cacheKey = `${cwd}\u0000${upstream.upstreamRef}`; - const now = Date.now(); - const lastRefreshAt = lastUpstreamRefreshAt.get(cacheKey) ?? 0; - if (now - lastRefreshAt < STATUS_UPSTREAM_REFRESH_INTERVAL_MS) { - return; - } - - // Record attempt time first to avoid repeated fetch spikes when status is polled often. - lastUpstreamRefreshAt.set(cacheKey, now); - yield* fetchUpstreamRefForStatus(cwd, upstream); + if (!upstream) return; + yield* Cache.get( + statusUpstreamRefreshCache, + new StatusUpstreamRefreshCacheKey({ + cwd, + upstreamRef: upstream.upstreamRef, + remoteName: upstream.remoteName, + upstreamBranch: upstream.upstreamBranch, + }), + ); }); const refreshCheckedOutBranchUpstream = (cwd: string): Effect.Effect => Effect.gen(function* () { const upstream = yield* resolveCurrentUpstream(cwd); - if (!upstream) { - return; - } + if (!upstream) return; yield* fetchUpstreamRef(cwd, upstream); }); @@ -394,8 +405,8 @@ const makeGitCore = Effect.gen(function* () { "diff", "--cached", "--name-status", - ]); - if (trimStdout(stagedSummary).length === 0) { + ]).pipe(Effect.map((stdout) => stdout.trim())); + if (stagedSummary.length === 0) { return null; } @@ -420,9 +431,11 @@ const makeGitCore = Effect.gen(function* () { args.push("-m", trimmedBody); } yield* runGit("GitCore.commit.commit", cwd, args); - const commitSha = trimStdout( - yield* runGitStdout("GitCore.commit.revParseHead", cwd, ["rev-parse", "HEAD"]), - ); + const commitSha = yield* runGitStdout("GitCore.commit.revParseHead", cwd, [ + "rev-parse", + "HEAD", + ]).pipe(Effect.map((stdout) => stdout.trim())); + return { commitSha }; }); @@ -491,21 +504,23 @@ const makeGitCore = Effect.gen(function* () { "Current branch has no upstream configured. Push with upstream first.", ); } - const beforeSha = trimStdout( - yield* runGitStdout( - "GitCore.pullCurrentBranch.beforeSha", - cwd, - ["rev-parse", "HEAD"], - true, - ), - ); + const beforeSha = yield* runGitStdout( + "GitCore.pullCurrentBranch.beforeSha", + cwd, + ["rev-parse", "HEAD"], + true, + ).pipe(Effect.map((stdout) => stdout.trim())); yield* executeGit("GitCore.pullCurrentBranch.pull", cwd, ["pull", "--ff-only"], { timeoutMs: 30_000, fallbackErrorMessage: "git pull failed", }); - const afterSha = trimStdout( - yield* runGitStdout("GitCore.pullCurrentBranch.afterSha", cwd, ["rev-parse", "HEAD"], true), - ); + const afterSha = yield* runGitStdout( + "GitCore.pullCurrentBranch.afterSha", + cwd, + ["rev-parse", "HEAD"], + true, + ).pipe(Effect.map((stdout) => stdout.trim())); + const refreshed = yield* statusDetails(cwd); return { status: beforeSha.length > 0 && beforeSha === afterSha ? "skipped_up_to_date" : "pulled", @@ -540,10 +555,8 @@ const makeGitCore = Effect.gen(function* () { const readConfigValue: GitCoreShape["readConfigValue"] = (cwd, key) => runGitStdout("GitCore.readConfigValue", cwd, ["config", "--get", key], true).pipe( - Effect.map((stdout) => { - const value = trimStdout(stdout); - return value.length > 0 ? value : null; - }), + Effect.map((stdout) => stdout.trim()), + Effect.map((trimmed) => (trimmed.length > 0 ? trimmed : null)), ); const listBranches: GitCoreShape["listBranches"] = (input) => diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 6be2c8af31d3c9a83d04b13cb523c922403da3e4..d3ec9b676b636a50984f4d54cb52dcb358883e84 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -8,7 +8,7 @@ import { type ThreadId, type TurnId, } from "@t3tools/contracts"; -import { Effect, Layer, Queue, Stream } from "effect"; +import { Cache, Duration, Effect, Layer, Option, Queue, Stream } from "effect"; import { ProviderService } from "../../provider/Services/ProviderService.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; @@ -76,46 +76,23 @@ const serverCommandId = (tag: string): CommandId => CommandId.makeUnsafe(`server:${tag}:${crypto.randomUUID()}`); const HANDLED_TURN_START_KEY_MAX = 10_000; -const HANDLED_TURN_START_KEY_TTL_MS = 30 * 60 * 1_000; - -function pruneHandledTurnStartKeys(handledKeys: Map, nowMs: number): void { - const staleBefore = nowMs - HANDLED_TURN_START_KEY_TTL_MS; - for (const [key, seenAt] of handledKeys) { - if (seenAt >= staleBefore) { - break; - } - handledKeys.delete(key); - } - - while (handledKeys.size > HANDLED_TURN_START_KEY_MAX) { - const oldestKey = handledKeys.keys().next().value; - if (!oldestKey) { - break; - } - handledKeys.delete(oldestKey); - } -} - -function hasHandledTurnStartRecently( - handledKeys: Map, - key: string, - nowMs = Date.now(), -): boolean { - const seenAt = handledKeys.get(key); - if (seenAt !== undefined && nowMs - seenAt <= HANDLED_TURN_START_KEY_TTL_MS) { - handledKeys.set(key, nowMs); - return true; - } - - handledKeys.set(key, nowMs); - pruneHandledTurnStartKeys(handledKeys, nowMs); - return false; -} +const HANDLED_TURN_START_KEY_TTL = Duration.minutes(30); const make = Effect.gen(function* () { const orchestrationEngine = yield* OrchestrationEngineService; const providerService = yield* ProviderService; - const handledTurnStartKeys = new Map(); + const handledTurnStartKeys = yield* Cache.make({ + capacity: HANDLED_TURN_START_KEY_MAX, + timeToLive: HANDLED_TURN_START_KEY_TTL, + lookup: () => Effect.succeed(true), + }); + + const hasHandledTurnStartRecently = (key: string) => + Cache.getOption(handledTurnStartKeys, key).pipe( + Effect.flatMap((cached) => + Cache.set(handledTurnStartKeys, key, true).pipe(Effect.as(Option.isSome(cached))), + ), + ); const appendProviderFailureActivity = (input: { readonly threadId: ThreadId; @@ -246,7 +223,7 @@ const make = Effect.gen(function* () { event: Extract, ) { const key = turnStartKeyForEvent(event); - if (hasHandledTurnStartRecently(handledTurnStartKeys, key)) { + if (yield* hasHandledTurnStartRecently(key)) { return; } diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 7715bbfbfa2f80b5f15644f3d10f081e6848b31b..e76a355217d5d054da107e9c66f39673f1bbe5b5 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -7,7 +7,7 @@ import { type ProviderRuntimeEvent, type ProviderSessionId, } from "@t3tools/contracts"; -import { Cause, Effect, Layer, Queue, Stream } from "effect"; +import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Stream } from "effect"; import { ProviderService } from "../../provider/Services/ProviderService.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; @@ -19,6 +19,8 @@ import { const providerTurnKey = (sessionId: ProviderSessionId, turnId: TurnId) => `${sessionId}:${turnId}`; const providerCommandId = (event: ProviderRuntimeEvent, tag: string): CommandId => CommandId.makeUnsafe(`provider:${event.eventId}:${tag}:${crypto.randomUUID()}`); +const TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY = 10_000; +const TURN_MESSAGE_IDS_BY_TURN_TTL = Duration.minutes(120); function toTurnId(value: string | undefined): TurnId | undefined { return value === undefined ? undefined : TurnId.makeUnsafe(value); @@ -128,54 +130,76 @@ const make = Effect.gen(function* () { const orchestrationEngine = yield* OrchestrationEngineService; const providerService = yield* ProviderService; - const turnMessageIdsByTurnKey = new Map>(); + const turnMessageIdsByTurnKey = yield* Cache.make>({ + capacity: TURN_MESSAGE_IDS_BY_TURN_CACHE_CAPACITY, + timeToLive: TURN_MESSAGE_IDS_BY_TURN_TTL, + lookup: () => Effect.succeed(new Set()), + }); const rememberAssistantMessageId = ( sessionId: ProviderSessionId, turnId: TurnId, messageId: MessageId, - ) => { - const key = providerTurnKey(sessionId, turnId); - const existingIds = turnMessageIdsByTurnKey.get(key); - if (existingIds) { - existingIds.add(messageId); - } else { - turnMessageIdsByTurnKey.set(key, new Set([messageId])); - } - }; + ) => + Cache.getOption(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId)).pipe( + Effect.flatMap((existingIds) => + Cache.set( + turnMessageIdsByTurnKey, + providerTurnKey(sessionId, turnId), + Option.match(existingIds, { + onNone: () => new Set([messageId]), + onSome: (ids) => { + const nextIds = new Set(ids); + nextIds.add(messageId); + return nextIds; + }, + }), + ), + ), + ); const forgetAssistantMessageId = ( sessionId: ProviderSessionId, turnId: TurnId, messageId: MessageId, - ) => { - const key = providerTurnKey(sessionId, turnId); - const existingIds = turnMessageIdsByTurnKey.get(key); - if (!existingIds) { - return; - } - existingIds.delete(messageId); - if (existingIds.size === 0) { - turnMessageIdsByTurnKey.delete(key); - } - }; + ) => + Cache.getOption(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId)).pipe( + Effect.flatMap((existingIds) => + Option.match(existingIds, { + onNone: () => Effect.void, + onSome: (ids) => { + const nextIds = new Set(ids); + nextIds.delete(messageId); + if (nextIds.size === 0) { + return Cache.invalidate(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId)); + } + return Cache.set(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId), nextIds); + }, + }), + ), + ); - const getAssistantMessageIdsForTurn = (sessionId: ProviderSessionId, turnId: TurnId) => { - return turnMessageIdsByTurnKey.get(providerTurnKey(sessionId, turnId)) ?? new Set(); - }; + const getAssistantMessageIdsForTurn = (sessionId: ProviderSessionId, turnId: TurnId) => + Cache.getOption(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId)).pipe( + Effect.map((existingIds) => + Option.getOrElse(existingIds, (): Set => new Set()), + ), + ); - const clearAssistantMessageIdsForTurn = (sessionId: ProviderSessionId, turnId: TurnId) => { - turnMessageIdsByTurnKey.delete(providerTurnKey(sessionId, turnId)); - }; + const clearAssistantMessageIdsForTurn = (sessionId: ProviderSessionId, turnId: TurnId) => + Cache.invalidate(turnMessageIdsByTurnKey, providerTurnKey(sessionId, turnId)); - const clearTurnStateForSession = (sessionId: ProviderSessionId) => { - const prefix = `${sessionId}:`; - for (const key of turnMessageIdsByTurnKey.keys()) { - if (key.startsWith(prefix)) { - turnMessageIdsByTurnKey.delete(key); - } - } - }; + const clearTurnStateForSession = (sessionId: ProviderSessionId) => + Effect.gen(function* () { + const prefix = `${sessionId}:`; + const turnKeys = Array.from(yield* Cache.keys(turnMessageIdsByTurnKey)); + yield* Effect.forEach( + turnKeys, + (key) => + key.startsWith(prefix) ? Cache.invalidate(turnMessageIdsByTurnKey, key) : Effect.void, + { concurrency: 1 }, + ).pipe(Effect.asVoid); + }); const processEvent = (event: ProviderRuntimeEvent) => Effect.gen(function* () { @@ -243,7 +267,7 @@ const make = Effect.gen(function* () { ); const turnId = toTurnId(event.turnId); if (turnId) { - rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId); + yield* rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId); } yield* orchestrationEngine.dispatch({ @@ -261,7 +285,7 @@ const make = Effect.gen(function* () { const assistantMessageId = MessageId.makeUnsafe(`assistant:${event.itemId}`); const turnId = toTurnId(event.turnId); if (turnId) { - rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId); + yield* rememberAssistantMessageId(event.sessionId, turnId, assistantMessageId); } yield* orchestrationEngine.dispatch({ @@ -274,14 +298,14 @@ const make = Effect.gen(function* () { }); if (turnId) { - forgetAssistantMessageId(event.sessionId, turnId, assistantMessageId); + yield* forgetAssistantMessageId(event.sessionId, turnId, assistantMessageId); } } if (event.type === "turn.completed") { const turnId = toTurnId(event.turnId); if (turnId) { - const assistantMessageIds = getAssistantMessageIdsForTurn(event.sessionId, turnId); + const assistantMessageIds = yield* getAssistantMessageIdsForTurn(event.sessionId, turnId); yield* Effect.forEach(assistantMessageIds, (assistantMessageId) => orchestrationEngine.dispatch({ type: "thread.message.assistant.complete", @@ -292,12 +316,12 @@ const make = Effect.gen(function* () { createdAt: now, }), ).pipe(Effect.asVoid); - clearAssistantMessageIdsForTurn(event.sessionId, turnId); + yield* clearAssistantMessageIdsForTurn(event.sessionId, turnId); } } if (event.type === "session.exited") { - clearTurnStateForSession(event.sessionId); + yield* clearTurnStateForSession(event.sessionId); } if (event.type === "runtime.error") { diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 7b504ef03076c10d52f12574c312c79c141cfb0d..1a50a95d22cc7d2164b2d91b21d611eab719c0b1 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -12,7 +12,6 @@ import { NonNegativeInt, ProviderSessionId, - ProviderThreadId, ThreadId, ProviderInterruptTurnInput, ProviderRespondToRequestInput, @@ -94,14 +93,6 @@ function toRuntimePayloadFromSession(session: ProviderSession): Record Effect.gen(function* () { const canonicalEventLogger = @@ -126,7 +117,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => if (liveSessionId === event.sessionId) { return { ...event, - sessionId: ProviderSessionId.makeUnsafe(staleSessionId), + sessionId: staleSessionId, } satisfies ProviderRuntimeEvent; } } @@ -162,9 +153,8 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => ); } - const providerSessionId = asProviderSessionId(session.sessionId); yield* directory.upsert({ - sessionId: providerSessionId, + sessionId: session.sessionId, provider: session.provider, threadId, providerThreadId, @@ -444,7 +434,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => sessionId: routed.sessionId, }); const threadId = yield* directory - .getThreadId(asProviderSessionId(input.sessionId)) + .getThreadId(input.sessionId) .pipe(Effect.map(Option.getOrUndefined)); if (!threadId) { return yield* toValidationError( @@ -453,10 +443,10 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => ); } yield* directory.upsert({ - sessionId: asProviderSessionId(input.sessionId), + sessionId: input.sessionId, provider: routed.adapter.provider, threadId, - providerThreadId: asProviderThreadId(turn.threadId), + providerThreadId: turn.threadId, status: "running", ...(turn.resumeCursor !== undefined ? { resumeCursor: turn.resumeCursor } : {}), runtimePayload: { @@ -514,10 +504,10 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => yield* routed.adapter.stopSession(routed.sessionId); } if (routed.sessionId !== input.sessionId) { - yield* directory.remove(asProviderSessionId(routed.sessionId)); + yield* directory.remove(routed.sessionId); yield* clearAliasesReferencing(routed.sessionId); } - yield* directory.remove(asProviderSessionId(input.sessionId)); + yield* directory.remove(input.sessionId); yield* clearAliasesReferencing(input.sessionId); }); -- 2.54.0