| @@ -1,10 +1,9 @@ |
| 1 | // @effect-diagnostics nodeBuiltinImport:off | 1 | // @effect-diagnostics nodeBuiltinImport:off |
| 2 | /** | 2 | /** |
| 3 | * Provider event logger helper. | 3 | * Best-effort provider event logging with one shared writer store. |
| 4 | * | 4 | * |
| 5 | * Best-effort writer for observability logs. Each record is formatted as a | 5 | * Native and canonical views share batching, rotation, and retention state so |
| 6 | * single effect-style text line in a thread-scoped file. Failures are | 6 | * they cannot race while appending to the same thread-scoped file. |
| 7 | * downgraded to warnings so provider runtime behavior is unaffected. | | |
| 8 | */ | 7 | */ |
| 9 | import * as NodeFS from "node:fs"; | 8 | import * as NodeFS from "node:fs"; |
| 10 | import * as NodePath from "node:path"; | 9 | import * as NodePath from "node:path"; |
| @@ -12,275 +11,592 @@ import * as NodePath from "node:path"; · 1 unmodified line |
| 12 | import type { ThreadId } from "@t3tools/contracts"; | 11 | import type { ThreadId } from "@t3tools/contracts"; |
| 13 | import { RotatingFileSink } from "@t3tools/shared/logging"; | 12 | import { RotatingFileSink } from "@t3tools/shared/logging"; |
| 14 | import { errorTag } from "@t3tools/shared/observability"; | 13 | import { errorTag } from "@t3tools/shared/observability"; |
| | 14 | import * as Clock from "effect/Clock"; |
| | 15 | import * as DateTime from "effect/DateTime"; |
| 15 | import * as Effect from "effect/Effect"; | 16 | import * as Effect from "effect/Effect"; |
| 16 | import * as Exit from "effect/Exit"; | 17 | import * as Exit from "effect/Exit"; |
| 17 | import * as Logger from "effect/Logger"; | | |
| 18 | import * as Schema from "effect/Schema"; | 18 | import * as Schema from "effect/Schema"; |
| 19 | import * as Scope from "effect/Scope"; | 19 | import * as Scope from "effect/Scope"; |
| 20 | import * as SynchronizedRef from "effect/SynchronizedRef"; | 20 | import * as SynchronizedRef from "effect/SynchronizedRef"; |
| 21 | | 21 | |
| 22 | import { toSafeThreadAttachmentSegment } from "../../attachmentStore.ts"; | 22 | import { toSafeThreadAttachmentSegment } from "../../attachmentStore.ts"; |
| 23 | | 23 | |
| 24 | const DEFAULT_MAX_BYTES = 10 * 1024 * 1024; | 24 | const MEBIBYTE = 1024 * 1024; |
| | 25 | const DAY_MS = 24 * 60 * 60 * 1_000; |
| | 26 | const MIN_RECORD_BYTES = 256; |
| | 27 | const DEFAULT_MAX_BYTES = 10 * MEBIBYTE; |
| 25 | const DEFAULT_MAX_FILES = 10; | 28 | const DEFAULT_MAX_FILES = 10; |
| 26 | const DEFAULT_BATCH_WINDOW_MS = 200; | 29 | const DEFAULT_BATCH_WINDOW_MS = 1_000; |
| | 30 | const DEFAULT_MAX_TOTAL_BYTES = 512 * MEBIBYTE; |
| | 31 | const DEFAULT_MAX_AGE_MS = 14 * DAY_MS; |
| | 32 | const DEFAULT_RETENTION_CHECK_INTERVAL_MS = 5 * 60 * 1_000; |
| | 33 | const DEFAULT_MAX_BUFFERED_BYTES = MEBIBYTE; |
| | 34 | const DEFAULT_MAX_BUFFERED_RECORDS = 512; |
| | 35 | const DEFAULT_MAX_RECORD_BYTES = MEBIBYTE; |
| | 36 | const DEFAULT_MAX_CACHED_SINKS = 64; |
| 27 | const GLOBAL_THREAD_SEGMENT = "_global"; | 37 | const GLOBAL_THREAD_SEGMENT = "_global"; |
| 28 | const LOG_SCOPE = "provider-observability"; | 38 | const LOG_SCOPE = "provider-observability"; |
| | 39 | const PROVIDER_LOG_FILE_PATTERN = /\.log(?:\.\d+)?$/u; |
| 29 | const encodeUnknownJsonString = Schema.encodeUnknownEffect(Schema.UnknownFromJsonString); | 40 | const encodeUnknownJsonString = Schema.encodeUnknownEffect(Schema.UnknownFromJsonString); |
| 30 | | 41 | |
| | 42 | const transientCanonicalEventTypes = new Set([ |
| | 43 | "content.delta", |
| | 44 | "hook.progress", |
| | 45 | "item.updated", |
| | 46 | "task.progress", |
| | 47 | "thread.realtime.audio.delta", |
| | 48 | "tool.progress", |
| | 49 | "turn.proposed.delta", |
| | 50 | ]); |
| | 51 | |
| 31 | export type EventNdjsonStream = "native" | "canonical" | "orchestration"; | 52 | export type EventNdjsonStream = "native" | "canonical" | "orchestration"; |
| 32 | | 53 | |
| 33 | export interface EventNdjsonLogger { | 54 | export interface EventNdjsonLogger { |
| 34 | readonly filePath: string; | 55 | readonly filePath: string; |
| 35 | write: (event: unknown, threadId: ThreadId | null) => Effect.Effect<void, never, never>; | 56 | readonly write: (event: unknown, threadId: ThreadId | null) => Effect.Effect<void>; |
| 36 | close: () => Effect.Effect<void, never, never>; | 57 | readonly close: () => Effect.Effect<void>; |
| 37 | } | 58 | } |
| 38 | | 59 | |
| 39 | export interface EventNdjsonLoggerOptions { | 60 | export interface EventNdjsonLogStore { |
| 40 | readonly stream: EventNdjsonStream; | 61 | readonly filePath: string; |
| | 62 | readonly logger: (stream: EventNdjsonStream) => EventNdjsonLogger; |
| | 63 | readonly close: () => Effect.Effect<void>; |
| | 64 | } |
| | 65 | |
| | 66 | export interface EventNdjsonLogStoreOptions { |
| 41 | readonly maxBytes?: number; | 67 | readonly maxBytes?: number; |
| 42 | readonly maxFiles?: number; | 68 | readonly maxFiles?: number; |
| 43 | readonly batchWindowMs?: number; | 69 | readonly batchWindowMs?: number; |
| | 70 | readonly maxTotalBytes?: number; |
| | 71 | readonly maxAgeMs?: number; |
| | 72 | readonly retentionCheckIntervalMs?: number; |
| | 73 | readonly maxBufferedBytes?: number; |
| | 74 | readonly maxBufferedRecords?: number; |
| | 75 | readonly maxRecordBytes?: number; |
| | 76 | readonly maxCachedSinks?: number; |
| | 77 | } |
| | 78 | |
| | 79 | export interface EventNdjsonLoggerOptions extends EventNdjsonLogStoreOptions { |
| | 80 | readonly stream: EventNdjsonStream; |
| | 81 | } |
| | 82 | |
| | 83 | export class EventNdjsonLogConfigurationError extends Schema.TaggedErrorClass<EventNdjsonLogConfigurationError>()( |
| | 84 | "EventNdjsonLogConfigurationError", |
| | 85 | { |
| | 86 | filePath: Schema.String, |
| | 87 | option: Schema.String, |
| | 88 | value: Schema.Number, |
| | 89 | minimum: Schema.Number, |
| | 90 | }, |
| | 91 | ) { |
| | 92 | override get message(): string { |
| | 93 | return `Provider event log option '${this.option}' must be an integer >= ${this.minimum}; received ${this.value} for '${this.filePath}'`; |
| | 94 | } |
| | 95 | } |
| | 96 | |
| | 97 | export class EventNdjsonLogDirectoryError extends Schema.TaggedErrorClass<EventNdjsonLogDirectoryError>()( |
| | 98 | "EventNdjsonLogDirectoryError", |
| | 99 | { |
| | 100 | directory: Schema.String, |
| | 101 | cause: Schema.Defect(), |
| | 102 | }, |
| | 103 | ) { |
| | 104 | override get message(): string { |
| | 105 | return `Failed to create provider event log directory '${this.directory}'`; |
| | 106 | } |
| 44 | } | 107 | } |
| 45 | | 108 | |
| 46 | interface ThreadWriter { | 109 | export type EventNdjsonLogStoreError = |
| 47 | writeMessage: (message: string) => Effect.Effect<void>; | 110 | | EventNdjsonLogConfigurationError |
| 48 | close: () => Effect.Effect<void>; | 111 | | EventNdjsonLogDirectoryError; |
| | 112 | |
| | 113 | interface ResolvedOptions { |
| | 114 | readonly maxBytes: number; |
| | 115 | readonly maxFiles: number; |
| | 116 | readonly batchWindowMs: number; |
| | 117 | readonly maxTotalBytes: number; |
| | 118 | readonly maxAgeMs: number; |
| | 119 | readonly retentionCheckIntervalMs: number; |
| | 120 | readonly maxBufferedBytes: number; |
| | 121 | readonly maxBufferedRecords: number; |
| | 122 | readonly maxRecordBytes: number; |
| | 123 | readonly maxCachedSinks: number; |
| 49 | } | 124 | } |
| 50 | | 125 | |
| 51 | interface LoggerState { | 126 | interface PendingRecord { |
| 52 | readonly threadWriters: Map<string, ThreadWriter>; | 127 | readonly threadSegment: string; |
| 53 | readonly failedSegments: Set<string>; | 128 | readonly line: string; |
| | 129 | readonly bytes: number; |
| 54 | } | 130 | } |
| 55 | | 131 | |
| 56 | function logWarning(message: string, context: Record<string, unknown>): Effect.Effect<void> { | 132 | interface StoreState { |
| | 133 | readonly pending: ReadonlyArray<PendingRecord>; |
| | 134 | readonly pendingBytes: number; |
| | 135 | readonly sinks: ReadonlyMap<string, RotatingFileSink>; |
| | 136 | readonly closed: boolean; |
| | 137 | readonly lastRetentionAt: number; |
| | 138 | readonly retainedBytesEstimate: number; |
| | 139 | } |
| | 140 | |
| | 141 | interface FileOperationFailure { |
| | 142 | readonly filePath: string; |
| | 143 | readonly cause: unknown; |
| | 144 | } |
| | 145 | |
| | 146 | interface RetentionResult { |
| | 147 | readonly failures: ReadonlyArray<FileOperationFailure>; |
| | 148 | readonly removedCount: number; |
| | 149 | readonly totalBytes: number; |
| | 150 | } |
| | 151 | |
| | 152 | interface DrainResult { |
| | 153 | readonly failures: ReadonlyArray<FileOperationFailure>; |
| | 154 | } |
| | 155 | |
| | 156 | function logWarning(message: string, context: Record<string, unknown>) { |
| 57 | return Effect.logWarning(message, context).pipe(Effect.annotateLogs({ scope: LOG_SCOPE })); | 157 | return Effect.logWarning(message, context).pipe(Effect.annotateLogs({ scope: LOG_SCOPE })); |
| 58 | } | 158 | } |
| 59 | | 159 | |
| 60 | function resolveThreadSegment(raw: string | null | undefined): string { | 160 | function resolveThreadSegment(raw: string | null | undefined) { |
| 61 | const normalized = typeof raw === "string" ? toSafeThreadAttachmentSegment(raw) : null; | 161 | const normalized = typeof raw === "string" ? toSafeThreadAttachmentSegment(raw) : null; |
| 62 | return normalized ?? GLOBAL_THREAD_SEGMENT; | 162 | return normalized ?? GLOBAL_THREAD_SEGMENT; |
| 63 | } | 163 | } |
| 64 | | 164 | |
| 65 | function formatLoggerMessage(message: unknown): string { | 165 | function resolveStreamLabel(stream: EventNdjsonStream) { |
| 66 | if (Array.isArray(message)) { | 166 | return stream === "native" ? "NTIVE" : "CANON"; |
| 67 | return message.map((part) => (typeof part === "string" ? part : String(part))).join(" "); | 167 | } |
| | 168 | |
| | 169 | function shouldPersist(stream: EventNdjsonStream, event: unknown) { |
| | 170 | if (stream !== "canonical" || typeof event !== "object" || event === null) { |
| | 171 | return true; |
| 68 | } | 172 | } |
| 69 | return typeof message === "string" ? message : String(message); | 173 | const type = Reflect.get(event, "type"); |
| | 174 | return typeof type !== "string" || !transientCanonicalEventTypes.has(type); |
| 70 | } | 175 | } |
| 71 | | 176 | |
| 72 | function makeLineLogger(streamLabel: string): Logger.Logger<unknown, string> { | 177 | function eventType(event: unknown) { |
| 73 | return Logger.make( | 178 | if (typeof event !== "object" || event === null) return null; |
| 74 | ({ date, message }) => | 179 | const type = Reflect.get(event, "type"); |
| 75 | `[${date.toISOString()}] ${streamLabel}: ${formatLoggerMessage(message)}\n`, | 180 | return typeof type === "string" ? type.slice(0, 128) : null; |
| 76 | ); | | |
| 77 | } | 181 | } |
| 78 | | 182 | |
| 79 | function resolveStreamLabel(stream: EventNdjsonStream): string { | 183 | function formatRecordLine(input: { |
| 80 | switch (stream) { | 184 | readonly stream: EventNdjsonStream; |
| 81 | case "native": | 185 | readonly event: unknown; |
| 82 | return "NTIVE"; | 186 | readonly observedAt: string; |
| 83 | case "canonical": | 187 | readonly payload: string; |
| 84 | case "orchestration": | 188 | readonly maxRecordBytes: number; |
| 85 | default: | 189 | }) { |
| 86 | return "CANON"; | 190 | const prefix = `[${input.observedAt}] ${resolveStreamLabel(input.stream)}: `; |
| | 191 | const line = `${prefix}${input.payload}\n`; |
| | 192 | const bytes = Buffer.byteLength(line); |
| | 193 | if (bytes <= input.maxRecordBytes) { |
| | 194 | return { line, bytes }; |
| | 195 | } |
| | 196 | |
| | 197 | const omittedPayload = (type: string | null) => |
| | 198 | JSON.stringify({ |
| | 199 | _tag: "ProviderEventLogRecordOmitted", |
| | 200 | reason: "record_too_large", |
| | 201 | originalBytes: bytes, |
| | 202 | eventType: type, |
| | 203 | }); |
| | 204 | let omittedLine = `${prefix}${omittedPayload(eventType(input.event))}\n`; |
| | 205 | if (Buffer.byteLength(omittedLine) > input.maxRecordBytes) { |
| | 206 | omittedLine = `${prefix}${omittedPayload(null)}\n`; |
| 87 | } | 207 | } |
| | 208 | return { |
| | 209 | line: omittedLine, |
| | 210 | bytes: Buffer.byteLength(omittedLine), |
| | 211 | }; |
| 88 | } | 212 | } |
| 89 | | 213 | |
| 90 | const toLogMessage = Effect.fn("toLogMessage")(function* ( | 214 | function writeBatchedRecords( |
| 91 | event: unknown, | 215 | sink: RotatingFileSink, |
| 92 | ): Effect.fn.Return<string | undefined> { | 216 | records: ReadonlyArray<PendingRecord>, |
| 93 | return yield* encodeUnknownJsonString(event).pipe( | 217 | maxChunkBytes: number, |
| 94 | Effect.catch((error) => | 218 | ) { |
| 95 | logWarning("failed to serialize provider event log record", { | 219 | let pendingLines: Array<string> = []; |
| 96 | errorTag: errorTag(error), | 220 | let pendingBytes = 0; |
| 97 | }).pipe(Effect.as(undefined)), | 221 | |
| 98 | ), | 222 | const flush = () => { |
| 99 | ); | 223 | if (pendingLines.length === 0) return; |
| 100 | }); | 224 | sink.write(pendingLines.join("")); |
| | 225 | pendingLines = []; |
| | 226 | pendingBytes = 0; |
| | 227 | }; |
| | 228 | |
| | 229 | for (const record of records) { |
| | 230 | if (pendingBytes > 0 && pendingBytes + record.bytes > maxChunkBytes) { |
| | 231 | flush(); |
| | 232 | } |
| | 233 | pendingLines.push(record.line); |
| | 234 | pendingBytes += record.bytes; |
| | 235 | if (pendingBytes >= maxChunkBytes) { |
| | 236 | flush(); |
| | 237 | } |
| | 238 | } |
| | 239 | flush(); |
| | 240 | } |
| 101 | | 241 | |
| 102 | const makeThreadWriter = Effect.fn("makeThreadWriter")(function* (input: { | 242 | function enforceRetention(input: { |
| 103 | readonly filePath: string; | 243 | readonly directory: string; |
| 104 | readonly maxBytes: number; | 244 | readonly maxTotalBytes: number; |
| 105 | readonly maxFiles: number; | 245 | readonly maxAgeMs: number; |
| 106 | readonly batchWindowMs: number; | 246 | readonly now: number; |
| 107 | readonly streamLabel: string; | 247 | }): RetentionResult { |
| 108 | }): Effect.fn.Return<ThreadWriter | undefined> { | 248 | const failures: Array<FileOperationFailure> = []; |
| 109 | const sinkResult = yield* Effect.sync(() => { | 249 | const files: Array<{ filePath: string; mtimeMs: number; size: number }> = []; |
| | 250 | |
| | 251 | let entries: ReadonlyArray<NodeFS.Dirent>; |
| | 252 | try { |
| | 253 | entries = NodeFS.readdirSync(input.directory, { withFileTypes: true }); |
| | 254 | } catch (cause) { |
| | 255 | return { |
| | 256 | failures: [{ filePath: input.directory, cause }], |
| | 257 | removedCount: 0, |
| | 258 | totalBytes: input.maxTotalBytes + 1, |
| | 259 | }; |
| | 260 | } |
| | 261 | |
| | 262 | for (const entry of entries) { |
| | 263 | if (!entry.isFile() || !PROVIDER_LOG_FILE_PATTERN.test(entry.name)) continue; |
| | 264 | const filePath = NodePath.join(input.directory, entry.name); |
| 110 | try { | 265 | try { |
| 111 | return { | 266 | const stat = NodeFS.statSync(filePath); |
| 112 | ok: true as const, | 267 | files.push({ filePath, mtimeMs: stat.mtimeMs, size: stat.size }); |
| 113 | sink: new RotatingFileSink({ | 268 | } catch (cause) { |
| 114 | filePath: input.filePath, | 269 | failures.push({ filePath, cause }); |
| 115 | maxBytes: input.maxBytes, | | |
| 116 | maxFiles: input.maxFiles, | | |
| 117 | throwOnError: true, | | |
| 118 | }), | | |
| 119 | }; | | |
| 120 | } catch (error) { | | |
| 121 | return { ok: false as const, error }; | | |
| 122 | } | 270 | } |
| | 271 | } |
| | 272 | |
| | 273 | let totalBytes = files.reduce((total, file) => total + file.size, 0); |
| | 274 | let removedCount = 0; |
| | 275 | const remove = (file: (typeof files)[number]) => { |
| | 276 | try { |
| | 277 | NodeFS.rmSync(file.filePath, { force: true }); |
| | 278 | totalBytes -= file.size; |
| | 279 | removedCount += 1; |
| | 280 | return true; |
| | 281 | } catch (cause) { |
| | 282 | failures.push({ filePath: file.filePath, cause }); |
| | 283 | return false; |
| | 284 | } |
| | 285 | }; |
| | 286 | |
| | 287 | const retained = files.filter((file) => { |
| | 288 | if (input.now - file.mtimeMs <= input.maxAgeMs) return true; |
| | 289 | return !remove(file); |
| 123 | }); | 290 | }); |
| 124 | | 291 | |
| 125 | if (!sinkResult.ok) { | 292 | for (const file of retained.toSorted( |
| 126 | yield* logWarning("failed to initialize provider thread log file", { | 293 | (left, right) => left.mtimeMs - right.mtimeMs || left.filePath.localeCompare(right.filePath), |
| 127 | filePath: input.filePath, | 294 | )) { |
| 128 | errorTag: errorTag(sinkResult.error), | 295 | if (totalBytes <= input.maxTotalBytes) break; |
| 129 | }); | 296 | remove(file); |
| 130 | return undefined; | | |
| 131 | } | 297 | } |
| 132 | | 298 | |
| 133 | const sink = sinkResult.sink; | 299 | return { failures, removedCount, totalBytes }; |
| 134 | const scope = yield* Scope.make(); | 300 | } |
| 135 | const lineLogger = makeLineLogger(input.streamLabel); | 301 | |
| 136 | const batchedLogger = yield* Logger.batched(lineLogger, { | 302 | function validateOption(input: { |
| 137 | window: input.batchWindowMs, | 303 | readonly filePath: string; |
| 138 | flush: Effect.fn("makeThreadWriter.flush")(function* (messages) { | 304 | readonly option: string; |
| 139 | const flushResult = yield* Effect.sync(() => { | 305 | readonly value: number; |
| 140 | try { | 306 | readonly minimum: number; |
| 141 | for (const message of messages) { | 307 | }) { |
| 142 | sink.write(message); | 308 | if (Number.isInteger(input.value) && input.value >= input.minimum) return undefined; |
| 143 | } | 309 | return new EventNdjsonLogConfigurationError(input); |
| 144 | return { ok: true as const }; | 310 | } |
| 145 | } catch (error) { | 311 | |
| 146 | return { ok: false as const, error }; | 312 | function resolveOptions(filePath: string, options: EventNdjsonLogStoreOptions) { |
| 147 | } | 313 | const maxBytes = options.maxBytes ?? DEFAULT_MAX_BYTES; |
| 148 | }); | 314 | const maxBufferedBytes = options.maxBufferedBytes ?? DEFAULT_MAX_BUFFERED_BYTES; |
| | 315 | const requestedMaxRecordBytes = options.maxRecordBytes ?? DEFAULT_MAX_RECORD_BYTES; |
| | 316 | const resolved = { |
| | 317 | maxBytes, |
| | 318 | maxFiles: options.maxFiles ?? DEFAULT_MAX_FILES, |
| | 319 | batchWindowMs: options.batchWindowMs ?? DEFAULT_BATCH_WINDOW_MS, |
| | 320 | maxTotalBytes: options.maxTotalBytes ?? DEFAULT_MAX_TOTAL_BYTES, |
| | 321 | maxAgeMs: options.maxAgeMs ?? DEFAULT_MAX_AGE_MS, |
| | 322 | retentionCheckIntervalMs: |
| | 323 | options.retentionCheckIntervalMs ?? DEFAULT_RETENTION_CHECK_INTERVAL_MS, |
| | 324 | maxBufferedBytes, |
| | 325 | maxBufferedRecords: options.maxBufferedRecords ?? DEFAULT_MAX_BUFFERED_RECORDS, |
| | 326 | maxRecordBytes: Math.min(requestedMaxRecordBytes, maxBytes, maxBufferedBytes), |
| | 327 | maxCachedSinks: options.maxCachedSinks ?? DEFAULT_MAX_CACHED_SINKS, |
| | 328 | } satisfies ResolvedOptions; |
| | 329 | |
| | 330 | const validations = [ |
| | 331 | ["maxBytes", maxBytes, MIN_RECORD_BYTES], |
| | 332 | ["maxFiles", resolved.maxFiles, 1], |
| | 333 | ["batchWindowMs", resolved.batchWindowMs, 0], |
| | 334 | ["maxTotalBytes", resolved.maxTotalBytes, 1], |
| | 335 | ["maxAgeMs", resolved.maxAgeMs, 1], |
| | 336 | ["retentionCheckIntervalMs", resolved.retentionCheckIntervalMs, 1], |
| | 337 | ["maxBufferedBytes", maxBufferedBytes, MIN_RECORD_BYTES], |
| | 338 | ["maxBufferedRecords", resolved.maxBufferedRecords, 1], |
| | 339 | ["maxRecordBytes", requestedMaxRecordBytes, MIN_RECORD_BYTES], |
| | 340 | ["maxCachedSinks", resolved.maxCachedSinks, 1], |
| | 341 | ] as const; |
| | 342 | |
| | 343 | for (const [option, value, minimum] of validations) { |
| | 344 | const error = validateOption({ filePath, option, value, minimum }); |
| | 345 | if (error) return Effect.fail(error); |
| | 346 | } |
| | 347 | return Effect.succeed(resolved); |
| | 348 | } |
| | 349 | |
| | 350 | function drainPending(input: { |
| | 351 | readonly directory: string; |
| | 352 | readonly options: ResolvedOptions; |
| | 353 | readonly state: StoreState; |
| | 354 | readonly now: number; |
| | 355 | readonly close: boolean; |
| | 356 | }): readonly [DrainResult, StoreState] { |
| | 357 | if (input.state.closed) { |
| | 358 | return [{ failures: [] }, input.state]; |
| | 359 | } |
| 149 | | 360 | |
| 150 | if (!flushResult.ok) { | 361 | let sinks = new Map(input.state.sinks); |
| 151 | yield* logWarning("provider event log batch flush failed", { | 362 | const failures: Array<FileOperationFailure> = []; |
| 152 | filePath: input.filePath, | 363 | const recordsBySegment = new Map<string, Array<PendingRecord>>(); |
| 153 | errorTag: errorTag(flushResult.error), | 364 | |
| | 365 | for (const record of input.state.pending) { |
| | 366 | const records = recordsBySegment.get(record.threadSegment) ?? []; |
| | 367 | records.push(record); |
| | 368 | recordsBySegment.set(record.threadSegment, records); |
| | 369 | } |
| | 370 | |
| | 371 | let writtenBytes = 0; |
| | 372 | for (const [threadSegment, records] of recordsBySegment) { |
| | 373 | const filePath = NodePath.join(input.directory, `${threadSegment}.log`); |
| | 374 | let sink = sinks.get(threadSegment); |
| | 375 | try { |
| | 376 | if (!sink) { |
| | 377 | sink = new RotatingFileSink({ |
| | 378 | filePath, |
| | 379 | maxBytes: input.options.maxBytes, |
| | 380 | maxFiles: input.options.maxFiles, |
| | 381 | throwOnError: true, |
| 154 | }); | 382 | }); |
| 155 | } | 383 | } |
| 156 | }), | 384 | sinks.delete(threadSegment); |
| 157 | }).pipe(Effect.provideService(Scope.Scope, scope)); | 385 | sinks.set(threadSegment, sink); |
| | 386 | while (sinks.size > input.options.maxCachedSinks) { |
| | 387 | const oldest = sinks.keys().next().value; |
| | 388 | if (oldest === undefined) break; |
| | 389 | sinks.delete(oldest); |
| | 390 | } |
| | 391 | writeBatchedRecords( |
| | 392 | sink, |
| | 393 | records, |
| | 394 | Math.min(input.options.maxBytes, input.options.maxBufferedBytes), |
| | 395 | ); |
| | 396 | writtenBytes += records.reduce((total, record) => total + record.bytes, 0); |
| | 397 | } catch (cause) { |
| | 398 | sinks.delete(threadSegment); |
| | 399 | failures.push({ filePath, cause }); |
| | 400 | } |
| | 401 | } |
| 158 | | 402 | |
| 159 | const loggerLayer = Logger.layer([batchedLogger], { mergeWithExisting: false }); | 403 | const retainedBytesEstimate = input.state.retainedBytesEstimate + writtenBytes; |
| | 404 | const retentionDue = |
| | 405 | input.close || |
| | 406 | failures.length > 0 || |
| | 407 | retainedBytesEstimate > input.options.maxTotalBytes || |
| | 408 | input.now - input.state.lastRetentionAt >= input.options.retentionCheckIntervalMs; |
| | 409 | const retention = retentionDue |
| | 410 | ? enforceRetention({ |
| | 411 | directory: input.directory, |
| | 412 | maxTotalBytes: input.options.maxTotalBytes, |
| | 413 | maxAgeMs: input.options.maxAgeMs, |
| | 414 | now: input.now, |
| | 415 | }) |
| | 416 | : null; |
| | 417 | if (retention && retention.removedCount > 0) { |
| | 418 | sinks = new Map(); |
| | 419 | } |
| 160 | | 420 | |
| 161 | return { | 421 | return [ |
| 162 | writeMessage(message: string) { | 422 | { |
| 163 | return Effect.log(message).pipe(Effect.provide(loggerLayer)); | 423 | failures: [...failures, ...(retention?.failures ?? [])], |
| 164 | }, | 424 | }, |
| 165 | close() { | 425 | { |
| 166 | return Scope.close(scope, Exit.void); | 426 | pending: [], |
| | 427 | pendingBytes: 0, |
| | 428 | sinks, |
| | 429 | closed: input.close, |
| | 430 | lastRetentionAt: retention ? input.now : input.state.lastRetentionAt, |
| | 431 | retainedBytesEstimate: retention?.totalBytes ?? retainedBytesEstimate, |
| 167 | }, | 432 | }, |
| 168 | } satisfies ThreadWriter; | 433 | ]; |
| | 434 | } |
| | 435 | |
| | 436 | const serializeEvent = Effect.fnUntraced(function* (event: unknown) { |
| | 437 | return yield* encodeUnknownJsonString(event).pipe( |
| | 438 | Effect.catch((error) => |
| | 439 | logWarning("failed to serialize provider event log record", { |
| | 440 | errorTag: errorTag(error), |
| | 441 | }).pipe(Effect.as(undefined)), |
| | 442 | ), |
| | 443 | ); |
| 169 | }); | 444 | }); |
| 170 | | 445 | |
| 171 | export const makeEventNdjsonLogger = Effect.fn("makeEventNdjsonLogger")(function* ( | 446 | export const makeEventNdjsonLogStore = Effect.fnUntraced(function* ( |
| 172 | filePath: string, | 447 | filePath: string, |
| 173 | options: EventNdjsonLoggerOptions, | 448 | options: EventNdjsonLogStoreOptions = {}, |
| 174 | ): Effect.fn.Return<EventNdjsonLogger | undefined> { | 449 | ): Effect.fn.Return<EventNdjsonLogStore, EventNdjsonLogStoreError> { |
| 175 | const maxBytes = options.maxBytes ?? DEFAULT_MAX_BYTES; | 450 | const resolved = yield* resolveOptions(filePath, options); |
| 176 | const maxFiles = options.maxFiles ?? DEFAULT_MAX_FILES; | 451 | const directory = NodePath.dirname(filePath); |
| 177 | const batchWindowMs = options.batchWindowMs ?? DEFAULT_BATCH_WINDOW_MS; | 452 | |
| 178 | const streamLabel = resolveStreamLabel(options.stream); | 453 | yield* Effect.try({ |
| 179 | | 454 | try: () => NodeFS.mkdirSync(directory, { recursive: true }), |
| 180 | const directoryReady = yield* Effect.sync(() => { | 455 | catch: (cause) => new EventNdjsonLogDirectoryError({ directory, cause }), |
| 181 | try { | | |
| 182 | NodeFS.mkdirSync(NodePath.dirname(filePath), { recursive: true }); | | |
| 183 | return true; | | |
| 184 | } catch (error) { | | |
| 185 | return { ok: false as const, error }; | | |
| 186 | } | | |
| 187 | }); | 456 | }); |
| 188 | if (directoryReady !== true) { | 457 | |
| 189 | yield* logWarning("failed to create provider event log directory", { | 458 | const initializedAt = yield* Clock.currentTimeMillis; |
| 190 | filePath, | 459 | const initialRetention = yield* Effect.sync(() => |
| 191 | errorTag: errorTag(directoryReady.error), | 460 | enforceRetention({ |
| | 461 | directory, |
| | 462 | maxTotalBytes: resolved.maxTotalBytes, |
| | 463 | maxAgeMs: resolved.maxAgeMs, |
| | 464 | now: initializedAt, |
| | 465 | }), |
| | 466 | ); |
| | 467 | for (const failure of initialRetention.failures) { |
| | 468 | yield* logWarning("provider event log retention failed", { |
| | 469 | filePath: failure.filePath, |
| | 470 | errorTag: errorTag(failure.cause), |
| 192 | }); | 471 | }); |
| 193 | return undefined; | | |
| 194 | } | 472 | } |
| 195 | | 473 | |
| 196 | const stateRef = yield* SynchronizedRef.make<LoggerState>({ | 474 | const stateRef = yield* SynchronizedRef.make<StoreState>({ |
| 197 | threadWriters: new Map(), | 475 | pending: [], |
| 198 | failedSegments: new Set(), | 476 | pendingBytes: 0, |
| | 477 | sinks: new Map(), |
| | 478 | closed: false, |
| | 479 | lastRetentionAt: initializedAt, |
| | 480 | retainedBytesEstimate: initialRetention.totalBytes, |
| 199 | }); | 481 | }); |
| | 482 | const timerScope = yield* Scope.make(); |
| 200 | | 483 | |
| 201 | const resolveThreadWriter = Effect.fn("resolveThreadWriter")(function* ( | 484 | const reportDrainResult = Effect.fnUntraced(function* (result: DrainResult) { |
| 202 | threadSegment: string, | 485 | for (const failure of result.failures) { |
| 203 | ): Effect.fn.Return<ThreadWriter | undefined> { | 486 | yield* logWarning("provider event log write or retention failed", { |
| 204 | return yield* SynchronizedRef.modifyEffect(stateRef, (state) => { | 487 | filePath: failure.filePath, |
| 205 | if (state.failedSegments.has(threadSegment)) { | 488 | errorTag: errorTag(failure.cause), |
| 206 | return Effect.succeed([undefined, state] as const); | 489 | }); |
| 207 | } | 490 | } |
| 208 | | 491 | }); |
| 209 | const existing = state.threadWriters.get(threadSegment); | | |
| 210 | if (existing) { | | |
| 211 | return Effect.succeed([existing, state] as const); | | |
| 212 | } | | |
| 213 | | 492 | |
| 214 | return makeThreadWriter({ | 493 | const flush = Effect.fnUntraced(function* (close: boolean) { |
| 215 | filePath: NodePath.join(NodePath.dirname(filePath), `${threadSegment}.log`), | 494 | const now = yield* Clock.currentTimeMillis; |
| 216 | maxBytes, | 495 | const result = yield* SynchronizedRef.modifyEffect(stateRef, (state) => |
| 217 | maxFiles, | 496 | Effect.sync(() => |
| 218 | batchWindowMs, | 497 | drainPending({ |
| 219 | streamLabel, | 498 | directory, |
| 220 | }).pipe( | 499 | options: resolved, |
| 221 | Effect.map((writer) => { | 500 | state, |
| 222 | if (!writer) { | 501 | now, |
| 223 | const nextFailedSegments = new Set(state.failedSegments); | 502 | close, |
| 224 | nextFailedSegments.add(threadSegment); | | |
| 225 | return [ | | |
| 226 | undefined, | | |
| 227 | { | | |
| 228 | ...state, | | |
| 229 | failedSegments: nextFailedSegments, | | |
| 230 | }, | | |
| 231 | ] as const; | | |
| 232 | } | | |
| 233 | | | |
| 234 | const nextThreadWriters = new Map(state.threadWriters); | | |
| 235 | nextThreadWriters.set(threadSegment, writer); | | |
| 236 | return [ | | |
| 237 | writer, | | |
| 238 | { | | |
| 239 | ...state, | | |
| 240 | threadWriters: nextThreadWriters, | | |
| 241 | }, | | |
| 242 | ] as const; | | |
| 243 | }), | 503 | }), |
| 244 | ); | 504 | ), |
| 245 | }); | 505 | ); |
| | 506 | yield* reportDrainResult(result); |
| 246 | }); | 507 | }); |
| 247 | | 508 | |
| 248 | const write = Effect.fn("write")(function* (event: unknown, threadId: ThreadId | null) { | 509 | if (resolved.batchWindowMs > 0) { |
| 249 | const threadSegment = resolveThreadSegment(threadId); | 510 | yield* Effect.forkIn( |
| 250 | const message = yield* toLogMessage(event); | 511 | Effect.forever(Effect.sleep(resolved.batchWindowMs).pipe(Effect.andThen(flush(false)))), |
| 251 | if (!message) { | 512 | timerScope, |
| 252 | return; | 513 | { startImmediately: true }, |
| 253 | } | 514 | ); |
| 254 | | 515 | } |
| 255 | const writer = yield* resolveThreadWriter(threadSegment); | | |
| 256 | if (!writer) { | | |
| 257 | return; | | |
| 258 | } | | |
| 259 | | 516 | |
| 260 | yield* writer.writeMessage(message); | 517 | const close = Effect.fnUntraced(function* () { |
| | 518 | yield* flush(true); |
| | 519 | yield* Scope.close(timerScope, Exit.void); |
| 261 | }); | 520 | }); |
| 262 | | 521 | |
| 263 | const close = Effect.fn("close")(function* () { | 522 | const loggerViews = new Map<EventNdjsonStream, EventNdjsonLogger>(); |
| 264 | yield* SynchronizedRef.modifyEffect(stateRef, (state) => | 523 | const logger = (stream: EventNdjsonStream): EventNdjsonLogger => { |
| 265 | Effect.gen(function* () { | 524 | const existing = loggerViews.get(stream); |
| 266 | for (const writer of state.threadWriters.values()) { | 525 | if (existing) return existing; |
| 267 | yield* writer.close(); | 526 | |
| | 527 | const write = Effect.fnUntraced(function* (event: unknown, threadId: ThreadId | null) { |
| | 528 | if (!shouldPersist(stream, event)) return; |
| | 529 | const payload = yield* serializeEvent(event); |
| | 530 | if (payload === undefined) return; |
| | 531 | |
| | 532 | const observedAt = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso)); |
| | 533 | const record = formatRecordLine({ |
| | 534 | stream, |
| | 535 | event, |
| | 536 | observedAt, |
| | 537 | payload, |
| | 538 | maxRecordBytes: resolved.maxRecordBytes, |
| | 539 | }); |
| | 540 | const now = yield* Clock.currentTimeMillis; |
| | 541 | const result = yield* SynchronizedRef.modifyEffect(stateRef, (state) => { |
| | 542 | if (state.closed) { |
| | 543 | return Effect.succeed([null, state] as const); |
| 268 | } | 544 | } |
| 269 | | 545 | const pending = [ |
| 270 | return [ | 546 | ...state.pending, |
| 271 | undefined, | | |
| 272 | { | 547 | { |
| 273 | threadWriters: new Map<string, ThreadWriter>(), | 548 | threadSegment: resolveThreadSegment(threadId), |
| 274 | failedSegments: new Set<string>(), | 549 | line: record.line, |
| | 550 | bytes: record.bytes, |
| 275 | }, | 551 | }, |
| 276 | ] as const; | 552 | ]; |
| 277 | }), | 553 | const pendingBytes = state.pendingBytes + record.bytes; |
| 278 | ); | 554 | const shouldFlush = |
| 279 | }); | 555 | resolved.batchWindowMs === 0 || |
| | 556 | pending.length >= resolved.maxBufferedRecords || |
| | 557 | pendingBytes >= resolved.maxBufferedBytes; |
| | 558 | const nextState = { |
| | 559 | ...state, |
| | 560 | pending, |
| | 561 | pendingBytes, |
| | 562 | }; |
| | 563 | if (!shouldFlush) { |
| | 564 | return Effect.succeed([null, nextState] as const); |
| | 565 | } |
| | 566 | return Effect.sync(() => { |
| | 567 | const [drainResult, drainedState] = drainPending({ |
| | 568 | directory, |
| | 569 | options: resolved, |
| | 570 | state: nextState, |
| | 571 | now, |
| | 572 | close: false, |
| | 573 | }); |
| | 574 | return [drainResult, drainedState] as const; |
| | 575 | }); |
| | 576 | }); |
| | 577 | if (result) { |
| | 578 | yield* reportDrainResult(result); |
| | 579 | } |
| | 580 | }); |
| 280 | | 581 | |
| 281 | return { | 582 | const view = { filePath, write, close } satisfies EventNdjsonLogger; |
| 282 | filePath, | 583 | loggerViews.set(stream, view); |
| 283 | write, | 584 | return view; |
| 284 | close, | 585 | }; |
| 285 | } satisfies EventNdjsonLogger; | 586 | |
| | 587 | return { filePath, logger, close } satisfies EventNdjsonLogStore; |
| | 588 | }); |
| | 589 | |
| | 590 | export const makeEventNdjsonLogger = Effect.fnUntraced(function* ( |
| | 591 | filePath: string, |
| | 592 | options: EventNdjsonLoggerOptions, |
| | 593 | ): Effect.fn.Return<EventNdjsonLogger | undefined> { |
| | 594 | const store = yield* makeEventNdjsonLogStore(filePath, options).pipe( |
| | 595 | Effect.catch((error) => |
| | 596 | logWarning(error.message, { error }).pipe( |
| | 597 | Effect.as<EventNdjsonLogStore | undefined>(undefined), |
| | 598 | ), |
| | 599 | ), |
| | 600 | ); |
| | 601 | return store?.logger(options.stream); |
| 286 | }); | 602 | }); |