| author |  Julius Marminge <julius0216@outlook.com> 2026-10-05 21:37:12-07:00 |
| committer |  GitHub <noreply@github.com> 2026-10-05 21:37:12-07:00 |
| log | eef36a67c37fa74905fc7feaa836e5a81daa8685 |
| tree | e1834215f8fe70289bea56e22786d620c4636e26 |
| parent | 3236d8ce73d06df91e30532d0a5fd03cb537de8e |
| signature | Signed by PGP key B5690EEEBB952194 |
fix(server): caches and ids are written atomically (#16242)
Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>8 files changed, 126 insertions(+), 13 deletions(-)
apps
server
apps/server/src/atomicWrite.test.ts+46
| ... | @@ -137,3 +137,49 @@ it.effect("surfaces an unreadable link instead of writing over it", () => | ... | @@ -137,3 +137,49 @@ it.effect("surfaces an unreadable link instead of writing over it", () => |
| 137 | ), | 137 | ), |
| 138 | ), | 138 | ), |
| 139 | ); | 139 | ); |
| 140 | |||
| 141 | it.effect("succeeds when the write lands but its temp directory cannot be removed", () => | ||
| 142 | Effect.gen(function* () { | ||
| 143 | const renamed: Array<string> = []; | ||
| 144 | const removed: Array<string> = []; | ||
| 145 | const fileSystem = FileSystem.layerNoop({ | ||
| 146 | // The target does not exist yet, so it is written in place. | ||
| 147 | readLink: (path) => | ||
| 148 | Effect.fail( | ||
| 149 | PlatformError.systemError({ | ||
| 150 | _tag: "NotFound", | ||
| 151 | module: "FileSystem", | ||
| 152 | method: "readLink", | ||
| 153 | pathOrDescriptor: path, | ||
| 154 | }), | ||
| 155 | ), | ||
| 156 | makeDirectory: () => Effect.void, | ||
| 157 | makeTempDirectory: () => Effect.succeed("/home/settings.json.abc123"), | ||
| 158 | writeFileString: () => Effect.void, | ||
| 159 | rename: (_from, to) => Effect.sync(() => void renamed.push(to)), | ||
| 160 | remove: (path) => | ||
| 161 | Effect.sync(() => void removed.push(path)).pipe( | ||
| 162 | Effect.andThen( | ||
| 163 | Effect.fail( | ||
| 164 | PlatformError.systemError({ | ||
| 165 | _tag: "PermissionDenied", | ||
| 166 | module: "FileSystem", | ||
| 167 | method: "remove", | ||
| 168 | pathOrDescriptor: path, | ||
| 169 | }), | ||
| 170 | ), | ||
| 171 | ), | ||
| 172 | ), | ||
| 173 | }); | ||
| 174 | |||
| 175 | const result = yield* Effect.exit( | ||
| 176 | writeFileStringAtomically({ filePath: "/home/settings.json", contents: "after" }).pipe( | ||
| 177 | Effect.provide(Layer.mergeAll(Path.layer, fileSystem)), | ||
| 178 | ), | ||
| 179 | ); | ||
| 180 | |||
| 181 | assert.deepStrictEqual(result, Exit.void); | ||
| 182 | assert.deepStrictEqual(renamed, ["/home/settings.json"]); | ||
| 183 | assert.deepStrictEqual(removed, ["/home/settings.json.abc123"]); | ||
| 184 | }), | ||
| 185 | ); |
apps/server/src/atomicWrite.ts+10-4
| ... | @@ -20,10 +20,16 @@ export const writeFileStringAtomically = (input: { | ... | @@ -20,10 +20,16 @@ export const writeFileStringAtomically = (input: { |
| 20 | const targetDirectory = path.dirname(targetPath); | 20 | const targetDirectory = path.dirname(targetPath); |
| 21 | 21 | ||
| 22 | yield* fs.makeDirectory(targetDirectory, { recursive: true }); | 22 | yield* fs.makeDirectory(targetDirectory, { recursive: true }); |
| 23 | const tempDirectory = yield* fs.makeTempDirectoryScoped({ | 23 | // The temp directory is cleanup, not part of the write: failing to remove |
| 24 | directory: targetDirectory, | 24 | // it (a virus scanner holding it on Windows) must not fail a write that |
| 25 | prefix: `${path.basename(targetPath)}.`, | 25 | // already landed. |
| 26 | }); | 26 | const tempDirectory = yield* Effect.acquireRelease( |
| 27 | fs.makeTempDirectory({ | ||
| 28 | directory: targetDirectory, | ||
| 29 | prefix: `${path.basename(targetPath)}.`, | ||
| 30 | }), | ||
| 31 | (directory) => fs.remove(directory, { recursive: true }).pipe(Effect.ignore({ log: true })), | ||
| 32 | ); | ||
| 27 | const tempPath = path.join(tempDirectory, "contents.tmp"); | 33 | const tempPath = path.join(tempDirectory, "contents.tmp"); |
| 28 | 34 | ||
| 29 | yield* fs.writeFileString(tempPath, input.contents); | 35 | yield* fs.writeFileString(tempPath, input.contents); |
apps/server/src/device/LocalDeviceHost.ts+6-1
| ... | @@ -47,6 +47,7 @@ import { HttpClient } from "effect/http"; | ... | @@ -47,6 +47,7 @@ import { HttpClient } from "effect/http"; |
| 47 | import * as ChildProcess from "effect/process/ChildProcess"; | 47 | import * as ChildProcess from "effect/process/ChildProcess"; |
| 48 | import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; | 48 | import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; |
| 49 | 49 | ||
| 50 | import { writeFileStringAtomically } from "../atomicWrite.ts"; | ||
| 50 | import * as ServerConfig from "../config.ts"; | 51 | import * as ServerConfig from "../config.ts"; |
| 51 | import * as ProcessRunner from "../processRunner.ts"; | 52 | import * as ProcessRunner from "../processRunner.ts"; |
| 52 | import * as DeviceHost from "./DeviceHost.ts"; | 53 | import * as DeviceHost from "./DeviceHost.ts"; |
| ... | @@ -354,7 +355,11 @@ export const make = Effect.fn("LocalDeviceHost.make")(function* () { | ... | @@ -354,7 +355,11 @@ export const make = Effect.fn("LocalDeviceHost.make")(function* () { |
| 354 | port: Number(new URL(hub.origin).port), | 355 | port: Number(new URL(hub.origin).port), |
| 355 | entryPath: hubTool.entryPath, | 356 | entryPath: hubTool.entryPath, |
| 356 | }).pipe( | 357 | }).pipe( |
| 357 | Effect.flatMap((json) => fs.writeFileString(hubStatePath(), json)), | 358 | Effect.flatMap((contents) => |
| 359 | writeFileStringAtomically({ filePath: hubStatePath(), contents }), | ||
| 360 | ), | ||
| 361 | Effect.provideService(FileSystem.FileSystem, fs), | ||
| 362 | Effect.provideService(Path.Path, path), | ||
| 358 | Effect.ignore, | 363 | Effect.ignore, |
| 359 | ); | 364 | ); |
| 360 | 365 |
apps/server/src/provider/ModelManifest.ts+4-1
| ... | @@ -30,6 +30,7 @@ import * as Schema from "effect/Schema"; | ... | @@ -30,6 +30,7 @@ import * as Schema from "effect/Schema"; |
| 30 | import * as Semaphore from "effect/Semaphore"; | 30 | import * as Semaphore from "effect/Semaphore"; |
| 31 | import { HttpClient, HttpClientResponse } from "effect/http"; | 31 | import { HttpClient, HttpClientResponse } from "effect/http"; |
| 32 | 32 | ||
| 33 | import { writeFileStringAtomically } from "../atomicWrite.ts"; | ||
| 33 | import { ServerConfig } from "../config.ts"; | 34 | import { ServerConfig } from "../config.ts"; |
| 34 | import * as ServerSettings from "../serverSettings.ts"; | 35 | import * as ServerSettings from "../serverSettings.ts"; |
| 35 | import { hasValidClaudeManifestAdapters } from "./ClaudeModelManifest.ts"; | 36 | import { hasValidClaudeManifestAdapters } from "./ClaudeModelManifest.ts"; |
| ... | @@ -411,7 +412,9 @@ export const make = Effect.gen(function* () { | ... | @@ -411,7 +412,9 @@ export const make = Effect.gen(function* () { |
| 411 | manifest = fetched; | 412 | manifest = fetched; |
| 412 | fetchedAtMs = now; | 413 | fetchedAtMs = now; |
| 413 | yield* encodeManifestCache({ fetchedAtMs: now, manifest: fetched }).pipe( | 414 | yield* encodeManifestCache({ fetchedAtMs: now, manifest: fetched }).pipe( |
| 414 | Effect.flatMap((serialized) => fileSystem.writeFileString(cachePath, serialized)), | 415 | Effect.flatMap((contents) => writeFileStringAtomically({ filePath: cachePath, contents })), |
| 416 | Effect.provideService(FileSystem.FileSystem, fileSystem), | ||
| 417 | Effect.provideService(Path.Path, path), | ||
| 415 | Effect.ignoreCause, | 418 | Effect.ignoreCause, |
| 416 | ); | 419 | ); |
| 417 | return manifest; | 420 | return manifest; |
apps/server/src/provider/OpenCodeServerLedger.ts+8-2
| ... | @@ -8,6 +8,7 @@ import * as Path from "effect/Path"; | ... | @@ -8,6 +8,7 @@ import * as Path from "effect/Path"; |
| 8 | import * as Schema from "effect/Schema"; | 8 | import * as Schema from "effect/Schema"; |
| 9 | import { ChildProcess, ChildProcessSpawner } from "effect/process"; | 9 | import { ChildProcess, ChildProcessSpawner } from "effect/process"; |
| 10 | 10 | ||
| 11 | import { writeFileStringAtomically } from "../atomicWrite.ts"; | ||
| 11 | import * as ServerConfig from "../config.ts"; | 12 | import * as ServerConfig from "../config.ts"; |
| 12 | import { signalProcessGroup } from "../process/processGroup.ts"; | 13 | import { signalProcessGroup } from "../process/processGroup.ts"; |
| 13 | 14 | ||
| ... | @@ -245,8 +246,13 @@ export const make = Effect.fn("OpenCodeServerLedger.make")(function* (input: { | ... | @@ -245,8 +246,13 @@ export const make = Effect.fn("OpenCodeServerLedger.make")(function* (input: { |
| 245 | stateDir: input.stateDir, | 246 | stateDir: input.stateDir, |
| 246 | owner, | 247 | owner, |
| 247 | }; | 248 | }; |
| 248 | yield* fs.makeDirectory(directory, { recursive: true }); | 249 | yield* writeFileStringAtomically({ |
| 249 | yield* fs.writeFileString(entryPath, yield* encodeEntry(entry)); | 250 | filePath: entryPath, |
| 251 | contents: yield* encodeEntry(entry), | ||
| 252 | }).pipe( | ||
| 253 | Effect.provideService(FileSystem.FileSystem, fs), | ||
| 254 | Effect.provideService(Path.Path, path), | ||
| 255 | ); | ||
| 250 | return fs | 256 | return fs |
| 251 | .remove(entryPath, { force: true }) | 257 | .remove(entryPath, { force: true }) |
| 252 | .pipe( | 258 | .pipe( |
apps/server/src/telemetry/Identify.test.ts+40
| ... | @@ -6,6 +6,7 @@ import * as FileSystem from "effect/FileSystem"; | ... | @@ -6,6 +6,7 @@ import * as FileSystem from "effect/FileSystem"; |
| 6 | import * as Layer from "effect/Layer"; | 6 | import * as Layer from "effect/Layer"; |
| 7 | import * as Logger from "effect/Logger"; | 7 | import * as Logger from "effect/Logger"; |
| 8 | import * as Path from "effect/Path"; | 8 | import * as Path from "effect/Path"; |
| 9 | import * as PlatformError from "effect/PlatformError"; | ||
| 9 | import * as References from "effect/References"; | 10 | import * as References from "effect/References"; |
| 10 | 11 | ||
| 11 | import * as ServerConfig from "../config.ts"; | 12 | import * as ServerConfig from "../config.ts"; |
| ... | @@ -57,6 +58,45 @@ it.layer(NodeServices.layer)("telemetry identity", (it) => { | ... | @@ -57,6 +58,45 @@ it.layer(NodeServices.layer)("telemetry identity", (it) => { |
| 57 | ), | 58 | ), |
| 58 | ); | 59 | ); |
| 59 | 60 | ||
| 61 | it.effect("leaves no torn anonymous id behind when its write fails midway", () => | ||
| 62 | Effect.gen(function* () { | ||
| 63 | const config = yield* ServerConfig.ServerConfig; | ||
| 64 | const fileSystem = yield* FileSystem.FileSystem; | ||
| 65 | const path = yield* Path.Path; | ||
| 66 | const tornWrites = FileSystem.FileSystem.of({ | ||
| 67 | ...fileSystem, | ||
| 68 | writeFileString: (filePath, data, options) => | ||
| 69 | fileSystem | ||
| 70 | .writeFileString(filePath, data.slice(0, Math.floor(data.length / 2)), options) | ||
| 71 | .pipe( | ||
| 72 | Effect.andThen( | ||
| 73 | Effect.fail( | ||
| 74 | PlatformError.systemError({ | ||
| 75 | _tag: "WriteZero", | ||
| 76 | module: "FileSystem", | ||
| 77 | method: "writeFileString", | ||
| 78 | pathOrDescriptor: filePath, | ||
| 79 | }), | ||
| 80 | ), | ||
| 81 | ), | ||
| 82 | ), | ||
| 83 | }); | ||
| 84 | |||
| 85 | const identifier = yield* Identify.getTelemetryIdentifierForHome( | ||
| 86 | path.join(config.baseDir, "home"), | ||
| 87 | ).pipe(Effect.provideService(FileSystem.FileSystem, tornWrites)); | ||
| 88 | |||
| 89 | assert.isNull(identifier); | ||
| 90 | assert.isFalse(yield* fileSystem.exists(config.anonymousIdPath)); | ||
| 91 | }).pipe( | ||
| 92 | Effect.provide( | ||
| 93 | ServerConfig.layerTest(process.cwd(), { | ||
| 94 | prefix: "t3-telemetry-identify-torn-", | ||
| 95 | }), | ||
| 96 | ), | ||
| 97 | ), | ||
| 98 | ); | ||
| 99 | |||
| 60 | it.effect.each([ | 100 | it.effect.each([ |
| 61 | { | 101 | { |
| 62 | login: "an API key", | 102 | login: "an API key", |
apps/server/src/telemetry/Identify.ts+2-1
| ... | @@ -8,6 +8,7 @@ import * as Path from "effect/Path"; | ... | @@ -8,6 +8,7 @@ import * as Path from "effect/Path"; |
| 8 | import * as PlatformError from "effect/PlatformError"; | 8 | import * as PlatformError from "effect/PlatformError"; |
| 9 | import * as Schema from "effect/Schema"; | 9 | import * as Schema from "effect/Schema"; |
| 10 | 10 | ||
| 11 | import { writeFileStringAtomically } from "../atomicWrite.ts"; | ||
| 11 | import * as ServerConfig from "../config.ts"; | 12 | import * as ServerConfig from "../config.ts"; |
| 12 | 13 | ||
| 13 | /** | 14 | /** |
| ... | @@ -240,7 +241,7 @@ const upsertAnonymousId = Effect.gen(function* () { | ... | @@ -240,7 +241,7 @@ const upsertAnonymousId = Effect.gen(function* () { |
| 240 | }), | 241 | }), |
| 241 | ), | 242 | ), |
| 242 | ); | 243 | ); |
| 243 | yield* fileSystem.writeFileString(anonymousIdPath, anonymousId).pipe( | 244 | yield* writeFileStringAtomically({ filePath: anonymousIdPath, contents: anonymousId }).pipe( |
| 244 | Effect.mapError( | 245 | Effect.mapError( |
| 245 | (cause) => | 246 | (cause) => |
| 246 | new TelemetryAnonymousIdPersistenceError({ | 247 | new TelemetryAnonymousIdPersistenceError({ |
apps/server/src/usage/UsageService.ts+10-4
| ... | @@ -44,6 +44,7 @@ import * as Schema from "effect/Schema"; | ... | @@ -44,6 +44,7 @@ import * as Schema from "effect/Schema"; |
| 44 | import * as Semaphore from "effect/Semaphore"; | 44 | import * as Semaphore from "effect/Semaphore"; |
| 45 | import { HttpClient, HttpClientResponse } from "effect/http"; | 45 | import { HttpClient, HttpClientResponse } from "effect/http"; |
| 46 | 46 | ||
| 47 | import { writeFileStringAtomically } from "../atomicWrite.ts"; | ||
| 47 | import * as ServerConfig from "../config.ts"; | 48 | import * as ServerConfig from "../config.ts"; |
| 48 | import { expandHomePath } from "../pathExpansion.ts"; | 49 | import { expandHomePath } from "../pathExpansion.ts"; |
| 49 | import * as ServerSettings from "../serverSettings.ts"; | 50 | import * as ServerSettings from "../serverSettings.ts"; |
| ... | @@ -204,6 +205,11 @@ export const make = Effect.gen(function* () { | ... | @@ -204,6 +205,11 @@ export const make = Effect.gen(function* () { |
| 204 | const ratesCachePath = path.join(config.stateDir, "usage-model-rates.json"); | 205 | const ratesCachePath = path.join(config.stateDir, "usage-model-rates.json"); |
| 205 | const scanCachePath = path.join(config.stateDir, SCAN_CACHE_FILE_NAME); | 206 | const scanCachePath = path.join(config.stateDir, SCAN_CACHE_FILE_NAME); |
| 206 | const legacyScanCachePath = path.join(config.stateDir, LEGACY_SCAN_CACHE_FILE_NAME); | 207 | const legacyScanCachePath = path.join(config.stateDir, LEGACY_SCAN_CACHE_FILE_NAME); |
| 208 | const writeCacheFile = (filePath: string, contents: string) => | ||
| 209 | writeFileStringAtomically({ filePath, contents }).pipe( | ||
| 210 | Effect.provideService(FileSystem.FileSystem, fileSystem), | ||
| 211 | Effect.provideService(Path.Path, path), | ||
| 212 | ); | ||
| 207 | let rates: RateTable = new Map(); | 213 | let rates: RateTable = new Map(); |
| 208 | let ratesFetchedAtMs: number | null = null; | 214 | let ratesFetchedAtMs: number | null = null; |
| 209 | let ratesStatus: UsagePricing["status"] = "unavailable"; | 215 | let ratesStatus: UsagePricing["status"] = "unavailable"; |
| ... | @@ -267,7 +273,7 @@ export const make = Effect.gen(function* () { | ... | @@ -267,7 +273,7 @@ export const make = Effect.gen(function* () { |
| 267 | ratesStatus = "fresh"; | 273 | ratesStatus = "fresh"; |
| 268 | 274 | ||
| 269 | yield* encodeRatesCache({ fetchedAtMs: now, document: fetched }).pipe( | 275 | yield* encodeRatesCache({ fetchedAtMs: now, document: fetched }).pipe( |
| 270 | Effect.flatMap((serialized) => fileSystem.writeFileString(ratesCachePath, serialized)), | 276 | Effect.flatMap((contents) => writeCacheFile(ratesCachePath, contents)), |
| 271 | Effect.ignoreCause, | 277 | Effect.ignoreCause, |
| 272 | ); | 278 | ); |
| 273 | }); | 279 | }); |
| ... | @@ -419,8 +425,8 @@ export const make = Effect.gen(function* () { | ... | @@ -419,8 +425,8 @@ export const make = Effect.gen(function* () { |
| 419 | ); | 425 | ); |
| 420 | 426 | ||
| 421 | const writeScanCache = makeScanCacheWriter(); | 427 | const writeScanCache = makeScanCacheWriter(); |
| 422 | // Scans with different windows can finish together; two writes interleaved | 428 | // Scans with different windows can finish together; serializing the writes |
| 423 | // in one file would corrupt it. | 429 | // keeps an older snapshot from landing after a newer one. |
| 424 | const persistLock = yield* Semaphore.make(1); | 430 | const persistLock = yield* Semaphore.make(1); |
| 425 | 431 | ||
| 426 | const persistScanCache = Effect.fn("UsageService.persistScanCache")(function* () { | 432 | const persistScanCache = Effect.fn("UsageService.persistScanCache")(function* () { |
| ... | @@ -432,7 +438,7 @@ export const make = Effect.gen(function* () { | ... | @@ -432,7 +438,7 @@ export const make = Effect.gen(function* () { |
| 432 | yield* Effect.sync(() => | 438 | yield* Effect.sync(() => |
| 433 | writeScanCache(fileCache, { sources: Object.fromEntries(sourceCache) }), | 439 | writeScanCache(fileCache, { sources: Object.fromEntries(sourceCache) }), |
| 434 | ).pipe( | 440 | ).pipe( |
| 435 | Effect.flatMap((serialized) => fileSystem.writeFileString(scanCachePath, serialized)), | 441 | Effect.flatMap((contents) => writeCacheFile(scanCachePath, contents)), |
| 436 | // A cache we cannot write is a slower next start, not a failed read. | 442 | // A cache we cannot write is a slower next start, not a failed read. |
| 437 | Effect.catchCause(() => | 443 | Effect.catchCause(() => |
| 438 | Effect.sync(() => { | 444 | Effect.sync(() => { |