From 74c63d5d417282c00580621428d50a6f8a4e6ee8 Mon Sep 17 00:00:00 2001 From: Louis Haftmann <30736553+LouisHaftmann@users.noreply.github.com> Date: Sun, 13 Sep 2026 13:48:45 +0200 Subject: [PATCH 1/3] feat(helm): eagerMerge config option Created with AI. Verified by a human. Co-Authored-By: Claude Fable 5.1 --- CONTEXT.md | 8 + ...s-opt-in-and-composes-server-side-on-s3.md | 9 + .../github-actions-cache-server/Chart.yaml | 2 +- .../templates/_helpers.tpl | 2 + .../github-actions-cache-server/values.yaml | 6 + lib/schemas.ts | 1 + lib/storage.ts | 341 ++++++++++++------ tests/eager-merge.test.ts | 87 +++++ tests/setup.ts | 1 + tests/storage-lifecycle.test.ts | 8 +- 10 files changed, 353 insertions(+), 112 deletions(-) create mode 100644 docs/adr/0009-eager-merge-is-opt-in-and-composes-server-side-on-s3.md create mode 100644 tests/eager-merge.test.ts diff --git a/CONTEXT.md b/CONTEXT.md index ea062df..ca18f3e 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -43,6 +43,14 @@ A segment of cache data stored before its merged representation has been created **Merge**: The creation of a cache's consolidated stored representation from its Parts. +**Eager Merge**: +A Merge started at upload completion instead of on first download. Opt-in via `EAGER_MERGE`. +_Avoid_: pre-merge, upfront merge + +**Server-side Merge**: +A Merge the storage backend performs by copying Parts into the merged object without their bytes passing through the server (S3 `UploadPartCopy`). Requires every Part to satisfy the backend's limits. +_Avoid_: server-side copy, compose + **Merge Lease**: A time-bound, fenced claim granting one worker authority to complete a Merge. diff --git a/docs/adr/0009-eager-merge-is-opt-in-and-composes-server-side-on-s3.md b/docs/adr/0009-eager-merge-is-opt-in-and-composes-server-side-on-s3.md new file mode 100644 index 0000000..df0895c --- /dev/null +++ b/docs/adr/0009-eager-merge-is-opt-in-and-composes-server-side-on-s3.md @@ -0,0 +1,9 @@ +# Eager Merge is opt-in and composes server-side on S3 + +Direct downloads need a merged object, and the Merge runs lazily on first download, so the first Cache Hit of every entry is proxied through the server even with `ENABLE_DIRECT_DOWNLOADS`. `EAGER_MERGE=true` runs the Merge right after upload completion instead. It stays off by default because lazy merging never pays merge traffic for entries that are never downloaded. + +Both paths share one merge runner. Merge Lease, renewal, lease-fenced completion, and rollback are identical, so eager and lazy merges race safely against each other and against `cleanup:merges`. Upload completion picks the strategy up front from the Part sizes it already lists to record `sizeBytes`. When the adapter offers `composeParts` and every Part satisfies its limits (on S3: every Part but the last at least 5 MiB, none above 5 GiB, at most 10,000 Parts), the Merge is a Server-side Merge: `CreateMultipartUpload`, one `UploadPartCopy` per Part, `CompleteMultipartUpload`, with no bytes passing through the server. Otherwise the existing streaming merge runs immediately. Buildx uploads 1 MiB blocks, so its entries always take the streaming path. A backend that rejects `UploadPartCopy` fails the Merge like any other merge failure; the entry then merges lazily on first download. + +This does not reopen ADR-0004. The 5 GiB cap it cites applies to `CopyObject` used as a promote step. `UploadPartCopy` writes `merged` directly under the same lease fence, and Parts are immutable, so a merger that lost its lease can only write identical bytes. + +Costs: every entry occupies twice its size until `cleanup:parts` removes the Parts, not only entries that were downloaded, and `sizeBytes` still counts Parts only. A worker killed mid-composition leaves an incomplete multipart upload in the bucket; configure an `AbortIncompleteMultipartUpload` lifecycle rule. `FinalizeCacheEntryUpload` waits only for the Merge Lease, not for the Merge. GCS `compose` is not implemented; GCS entries take the streaming path. diff --git a/install/kubernetes/github-actions-cache-server/Chart.yaml b/install/kubernetes/github-actions-cache-server/Chart.yaml index 637ceee..d39bb2c 100644 --- a/install/kubernetes/github-actions-cache-server/Chart.yaml +++ b/install/kubernetes/github-actions-cache-server/Chart.yaml @@ -15,7 +15,7 @@ type: application # This is the chart version. This version number should be incremented each time you make changes # to the chart and its templates, including the app version. # Versions are expected to follow Semantic Versioning (https://semver.org/) -version: 1.3.0 +version: 1.4.0 # This is the version number of the application being deployed. This version number should be # incremented each time you make changes to the application. Versions are not expected to diff --git a/install/kubernetes/github-actions-cache-server/templates/_helpers.tpl b/install/kubernetes/github-actions-cache-server/templates/_helpers.tpl index f480c9b..ba685f8 100644 --- a/install/kubernetes/github-actions-cache-server/templates/_helpers.tpl +++ b/install/kubernetes/github-actions-cache-server/templates/_helpers.tpl @@ -125,6 +125,8 @@ Generate environment variables from config values. value: {{ default (printf "http://%s.%s.svc.cluster.local:%v" (include "github-actions-cache-server.fullname" .) .Release.Namespace .Values.service.port) .Values.config.apiBaseUrl | quote }} - name: ENABLE_DIRECT_DOWNLOADS value: {{ .Values.config.enableDirectDownloads | quote }} +- name: EAGER_MERGE + value: {{ .Values.config.eagerMerge | quote }} - name: CACHE_CLEANUP_OLDER_THAN_DAYS value: {{ .Values.config.cacheCleanupOlderThanDays | quote }} {{- if .Values.config.cacheMaxSizeBytes }} diff --git a/install/kubernetes/github-actions-cache-server/values.yaml b/install/kubernetes/github-actions-cache-server/values.yaml index 6e3dfa8..c544a7a 100644 --- a/install/kubernetes/github-actions-cache-server/values.yaml +++ b/install/kubernetes/github-actions-cache-server/values.yaml @@ -39,6 +39,12 @@ config: # The runner must be able to reach the storage provider directly. enableDirectDownloads: false + # -- Merge cache parts right after upload instead of on first download, so + # the first restore can be a direct download. On S3 the merge happens inside + # the bucket (UploadPartCopy) when every part but the last is at least 5 MiB. + # Doubles storage per entry until the parts cleanup job runs. + eagerMerge: false + # -- Number of days to keep stale cache data before deleting it. Set to 0 to disable. cacheCleanupOlderThanDays: 90 diff --git a/lib/schemas.ts b/lib/schemas.ts index bca21c6..27510ae 100644 --- a/lib/schemas.ts +++ b/lib/schemas.ts @@ -69,6 +69,7 @@ export const envBaseSchema = type({ 'DISABLE_CLEANUP_JOBS?': 'boolean', 'DEBUG?': 'unknown', 'ENABLE_DIRECT_DOWNLOADS': 'boolean = false', + 'EAGER_MERGE': 'boolean = false', 'BENCHMARK': 'boolean = false', 'SKIP_TOKEN_VALIDATION': 'boolean = false', 'MANAGEMENT_API_KEY?': 'string', diff --git a/lib/storage.ts b/lib/storage.ts index c31bd21..c75e937 100644 --- a/lib/storage.ts +++ b/lib/storage.ts @@ -13,19 +13,23 @@ import { PassThrough, Readable } from 'node:stream' import { pipeline } from 'node:stream/promises' import { createSingletonPromise } from '@antfu/utils' import { + AbortMultipartUploadCommand, + CompleteMultipartUploadCommand, + CreateMultipartUploadCommand, DeleteObjectsCommand, GetObjectCommand, HeadBucketCommand, HeadObjectCommand, ListObjectsV2Command, S3Client, + UploadPartCopyCommand, } from '@aws-sdk/client-s3' import { Upload as S3Upload } from '@aws-sdk/lib-storage' import { getSignedUrl } from '@aws-sdk/s3-request-presigner' import { Storage as GcsClient } from '@google-cloud/storage' import { NodeHttpHandler } from '@smithy/node-http-handler' import { sql } from 'kysely' -import { chunk } from 'remeda' +import { chunk, range } from 'remeda' import { match } from 'ts-pattern' import { getDatabase, retryOnLockConflict } from './db' import { env } from './env' @@ -220,6 +224,102 @@ export class Storage { } } + /** + * Runs a Merge under a Merge Lease. `write` produces `${folderName}/merged`; + * a lease-fenced transaction then marks the Merge complete. Returns nothing + * when another worker holds the lease. Writing straight to the final object + * is safe (ADR-0004). The returned promise never rejects: failures are logged + * and the merge state rolled back. + */ + private async startMerge(location: StorageLocation, write: () => Promise) { + const mergeToken = await acquireMergeLease(this.db, location.id) + if (!mergeToken) return + + await this.db + .updateTable('storage_locations') + .set({ mergeStartedAt: Date.now() }) + .where('id', '=', location.id) + .execute() + + const renewalTimer = setInterval(() => { + void renewMergeLease(this.db, location.id, mergeToken) + }, LEASE_RENEWAL_MS) + renewalTimer.unref() + + const mergePromise = write() + .then(async () => { + // The merged object is already written, so losing a deadlock here must + // not throw the merge away — the fence is re-checked on every attempt. + await retryOnLockConflict(() => + this.db.transaction().execute(async (tx) => { + let leaseQuery = tx + .selectFrom('merge_leases') + .select(['token', 'expiresAt']) + .where('storageLocationId', '=', location.id) + if (env.DB_DRIVER !== 'sqlite') leaseQuery = leaseQuery.forUpdate() + const lease = await leaseQuery.executeTakeFirst() + if (lease?.token !== mergeToken || lease.expiresAt <= Date.now()) + throw new Error('Merge lease was lost before completion') + await tx + .updateTable('storage_locations') + .set({ mergedAt: Date.now() }) + .where('id', '=', location.id) + .execute() + }), + ) + }) + .catch(async (err) => { + logger.error(`Merge failed for storage location ${location.id}`, { error: err }) + await this.db + .updateTable('storage_locations') + .set({ mergedAt: null, mergeStartedAt: null }) + .where('id', '=', location.id) + .where((eb) => + eb.exists( + eb + .selectFrom('merge_leases') + .select('storageLocationId') + .whereRef('storageLocationId', '=', 'storage_locations.id') + .where('token', '=', mergeToken), + ), + ) + .execute() + }) + .finally(async () => { + clearInterval(renewalTimer) + await releaseMergeLease(this.db, location.id, mergeToken) + }) + this.mergeStreamPromises.add(mergePromise) + mergePromise.finally(() => this.mergeStreamPromises.delete(mergePromise)) + return true + } + + /** + * Eager Merge (ADR-0009): a Server-side Merge when the adapter can compose + * these Parts, otherwise the streaming merge right away. Only the lease + * acquisition is awaited; the Merge itself runs in the background. + */ + private async mergeEagerly(location: StorageLocation, partSizes: (number | undefined)[]) { + const compose = this.adapter.composeParts + const composable = + compose && + partSizes.length <= compose.maxParts && + partSizes.every( + (bytes, index) => + bytes !== undefined && + bytes <= compose.maxPartBytes && + (index === partSizes.length - 1 || bytes >= compose.minPartBytes), + ) + await this.startMerge(location, () => + composable + ? compose.run(location.folderName, location.partCount) + : this.adapter.uploadStream( + `${location.folderName}/merged`, + Readable.from(this.streamParts(location)), + ), + ) + } + waitForOngoingMerges() { return Promise.all(this.mergeStreamPromises) } @@ -294,7 +394,8 @@ export class Storage { ) } - const partCount = await this.adapter.countFilesInFolder(`${upload.folderName}/parts`) + const parts = await this.adapter.listFolder(`${upload.folderName}/parts`) + const partCount = parts.length if (partCount !== upload.finishedPartUploadCount) { return this.abandonUpload( upload.id, @@ -305,23 +406,22 @@ export class Storage { ) } - const sizeBytes = await this.adapter.getFolderSize(upload.folderName) + const sizeByIndex = new Map(parts.map(({ name, bytes }) => [Number(name), bytes])) + const partSizes = Array.from({ length: partCount }, (_, index) => sizeByIndex.get(index)) + const location: StorageLocation = { + id: randomUUID(), + folderName: upload.folderName, + partCount, + mergedAt: null, + mergeStartedAt: null, + partsDeletedAt: null, + lastDownloadedAt: null, + sizeBytes: parts.reduce((sum, { bytes }) => sum + bytes, 0), + } + const locationId = location.id await this.db.transaction().execute(async (tx) => { - const locationId = randomUUID() - await tx - .insertInto('storage_locations') - .values({ - id: locationId, - folderName: upload.folderName, - partCount, - mergedAt: null, - mergeStartedAt: null, - partsDeletedAt: null, - lastDownloadedAt: null, - sizeBytes, - }) - .execute() + await tx.insertInto('storage_locations').values(location).execute() const existingCacheEntry = await tx .selectFrom('cache_entries') @@ -365,6 +465,14 @@ export class Storage { logger.warn('Capacity-based Eviction failed after upload completion', { error: err }) } + if (env.EAGER_MERGE) { + try { + await this.mergeEagerly(location, partSizes) + } catch (err) { + logger.warn('Eager Merge failed to start after upload completion', { error: err }) + } + } + return upload } @@ -447,82 +555,21 @@ export class Storage { await this.ensurePartsExist(storageLocation) - const mergeToken = await acquireMergeLease(this.db, storageLocation.id) - if (!mergeToken) { + const responseStream = new PassThrough() + const mergerStream = new PassThrough() + const merge = await this.startMerge(storageLocation, () => + this.adapter + .uploadStream(`${storageLocation.folderName}/merged`, mergerStream) + .catch((err) => { + mergerStream.destroy() + throw err + }), + ) + if (!merge) { const stream = await this.downloadFromCacheEntryLocation(storageLocation) return this.protectDownloadStream(stream, readerLeaseId) } - await this.db - .updateTable('storage_locations') - .set({ - mergeStartedAt: Date.now(), - }) - .where('id', '=', storageLocation.id) - .execute() - - const responseStream = new PassThrough() - const mergerStream = new PassThrough() - const renewalTimer = setInterval(() => { - void renewMergeLease(this.db, storageLocation.id, mergeToken) - }, LEASE_RENEWAL_MS) - renewalTimer.unref() - - // Uploading straight to the final object is safe: uploads are atomically - // visible (see StorageAdapter.uploadStream) and Parts are immutable, so a - // merger that lost its lease can only overwrite `merged` with identical - // bytes — the fence only needs to guard who flips `mergedAt`. - const mergePromise = this.adapter - .uploadStream(`${storageLocation.folderName}/merged`, mergerStream) - .then(async () => { - // The merged object is already written, so losing a deadlock here must - // not throw the merge away — the fence is re-checked on every attempt. - await retryOnLockConflict(() => - this.db.transaction().execute(async (tx) => { - let leaseQuery = tx - .selectFrom('merge_leases') - .select(['token', 'expiresAt']) - .where('storageLocationId', '=', storageLocation.id) - if (env.DB_DRIVER !== 'sqlite') leaseQuery = leaseQuery.forUpdate() - const lease = await leaseQuery.executeTakeFirst() - if (lease?.token !== mergeToken || lease.expiresAt <= Date.now()) - throw new Error('Merge lease was lost before completion') - await tx - .updateTable('storage_locations') - .set({ mergedAt: Date.now() }) - .where('id', '=', storageLocation.id) - .execute() - }), - ) - }) - .catch(async (err) => { - logger.error(`Merge failed for storage location ${storageLocation.id}`, { error: err }) - await this.db - .updateTable('storage_locations') - .set({ - mergedAt: null, - mergeStartedAt: null, - }) - .where('id', '=', storageLocation.id) - .where((eb) => - eb.exists( - eb - .selectFrom('merge_leases') - .select('storageLocationId') - .whereRef('storageLocationId', '=', 'storage_locations.id') - .where('token', '=', mergeToken), - ), - ) - .execute() - mergerStream.destroy() - }) - .finally(async () => { - clearInterval(renewalTimer) - await releaseMergeLease(this.db, storageLocation.id, mergeToken) - }) - this.mergeStreamPromises.add(mergePromise) - mergePromise.finally(() => this.mergeStreamPromises.delete(mergePromise)) - this.pumpPartsToStreams(storageLocation, responseStream, mergerStream).catch((err) => { responseStream.destroy(err) mergerStream.destroy(err) @@ -751,13 +798,31 @@ export interface StorageAdapter { objectExists(objectName: string): Promise deleteFolder(folderName: string): Promise countFilesInFolder(folderName: string): Promise - getFolderSize(folderName: string): Promise + /** Direct children of a folder, names relative to it. */ + listFolder(folderName: string): Promise listStorageFolders(): Promise createDownloadUrl?(objectName: string, expiresAt: number): Promise getFilesystemUsage?(): Promise<{ capacityBytes: number; usedBytes: number }> + /** + * Server-side Merge: copies `${folderName}/parts/0..partCount-1` into + * `${folderName}/merged` inside the backend. Only applicable when every Part + * satisfies the limits; the caller checks them before calling `run`. + */ + composeParts?: { + /** Every Part but the last must be at least this large. */ + minPartBytes: number + maxPartBytes: number + maxParts: number + run(folderName: string, partCount: number): Promise + } clear(): Promise } +export interface StorageObject { + name: string + bytes: number +} + export interface StorageFolder { folderName: string objectCount: number @@ -820,6 +885,52 @@ class S3Adapter implements StorageAdapter { private bucket private keyPrefix = 'gh-actions-cache' + // S3 multipart limits: https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html + composeParts = { + minPartBytes: 5 * 1024 * 1024, + maxPartBytes: 5 * 1024 ** 3, + maxParts: 10_000, + run: async (folderName: string, partCount: number) => { + const Bucket = this.bucket + const Key = `${this.keyPrefix}/${folderName}/merged` + const { UploadId } = await this.s3.send(new CreateMultipartUploadCommand({ Bucket, Key })) + if (!UploadId) throw new Error('S3 did not return an UploadId') + try { + const Parts = [] + const batches = chunk(range(0, partCount), 8) + for (const indexes of batches) { + const copies = await Promise.all( + indexes.map((index) => + this.s3.send( + new UploadPartCopyCommand({ + Bucket, + Key, + UploadId, + PartNumber: index + 1, + CopySource: `${Bucket}/${this.keyPrefix}/${folderName}/parts/${index}`, + }), + ), + ), + ) + Parts.push( + ...copies.map((copy, offset) => ({ + PartNumber: indexes[offset]! + 1, + ETag: copy.CopyPartResult?.ETag, + })), + ) + } + await this.s3.send( + new CompleteMultipartUploadCommand({ Bucket, Key, UploadId, MultipartUpload: { Parts } }), + ) + } catch (err) { + await this.s3 + .send(new AbortMultipartUploadCommand({ Bucket, Key, UploadId })) + .catch(() => {}) + throw err + } + }, + } + constructor({ bucket, s3 }: { s3: S3Client; bucket: string }) { this.s3 = s3 this.bucket = bucket @@ -960,14 +1071,16 @@ class S3Adapter implements StorageAdapter { return count } - async getFolderSize(folderName: string) { - let bytes = 0 - - const pages = this.listObjectsByPrefix(`${this.keyPrefix}/${folderName}/`) - for await (const listResponse of pages) - bytes += (listResponse.Contents ?? []).reduce((sum, object) => sum + (object.Size ?? 0), 0) - - return bytes + async listFolder(folderName: string) { + const prefix = `${this.keyPrefix}/${folderName}/` + const objects: StorageObject[] = [] + for await (const page of this.listObjectsByPrefix(prefix)) { + const contents = page.Contents ?? [] + for (const object of contents) + if (object.Key) + objects.push({ name: object.Key.slice(prefix.length), bytes: object.Size ?? 0 }) + } + return objects } async listStorageFolders() { @@ -1131,9 +1244,22 @@ class FileSystemAdapter implements StorageAdapter { } } - async getFolderSize(folderName: string) { - const folder = await this.inspectPath(this.safePath(folderName), folderName) - return folder.bytes + async listFolder(folderName: string) { + const folderPath = this.safePath(folderName) + let entries + try { + entries = await fs.readdir(folderPath, { withFileTypes: true }) + } catch (err: any) { + if (err.code === 'ENOENT') return [] + throw err + } + const files = entries.filter((entry) => entry.isFile()) + return Promise.all( + files.map(async (entry) => { + const stat = await fs.stat(path.join(folderPath, entry.name)) + return { name: entry.name, bytes: stat.size } + }), + ) } async listStorageFolders() { @@ -1219,12 +1345,13 @@ class GcsAdapter implements StorageAdapter { .then((res) => res[0].length) } - async getFolderSize(folderName: string) { - const [files] = await this.bucket.getFiles({ - prefix: `${this.keyPrefix}/${folderName}/`, - autoPaginate: true, - }) - return files.reduce((total, file) => total + Number(file.metadata.size ?? 0), 0) + async listFolder(folderName: string) { + const prefix = `${this.keyPrefix}/${folderName}/` + const [files] = await this.bucket.getFiles({ prefix, autoPaginate: true }) + return files.map((file) => ({ + name: file.name.slice(prefix.length), + bytes: Number(file.metadata.size ?? 0), + })) } async listStorageFolders() { diff --git a/tests/eager-merge.test.ts b/tests/eager-merge.test.ts new file mode 100644 index 0000000..deb2e81 --- /dev/null +++ b/tests/eager-merge.test.ts @@ -0,0 +1,87 @@ +import type { ReadableStream as NodeReadableStream } from 'node:stream/web' +import { randomUUID } from 'node:crypto' +import { Readable } from 'node:stream' + +import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest' +import { getDatabase } from '~/lib/db' +import { env } from '~/lib/env' +import { Storage } from '~/lib/storage' + +const scope = { version: 'v1', scope: 'refs/heads/main', repoId: '123' } + +async function uploadParts(storage: Storage, parts: Buffer[]) { + const key = randomUUID() + const upload = await storage.createUpload({ key, ...scope }) + for (const [index, part] of parts.entries()) + await storage.uploadPart( + upload!.id, + index, + Readable.toWeb(Readable.from(part)) as NodeReadableStream, + ) + await storage.completeUpload({ key, ...scope }) + const db = await getDatabase() + return db + .selectFrom('storage_locations') + .innerJoin('cache_entries', 'cache_entries.locationId', 'storage_locations.id') + .where('cache_entries.key', '=', key) + .selectAll('storage_locations') + .executeTakeFirstOrThrow() +} + +async function mergedBytes(storage: Storage, folderName: string) { + const stream = await storage.adapter.createDownloadStream(`${folderName}/merged`) + return Buffer.concat(await stream.toArray()) +} + +describe('eager merge', () => { + const originalEagerMerge = env.EAGER_MERGE + beforeEach(() => { + env.EAGER_MERGE = true + }) + afterEach(() => { + env.EAGER_MERGE = originalEagerMerge + }) + + test('streams parts into the merged object at upload completion', async () => { + const storage = await Storage.fromEnv() + const parts = [Buffer.alloc(1024, 'a'), Buffer.alloc(1024, 'b')] + + const location = await uploadParts(storage, parts) + try { + await storage.waitForOngoingMerges() + const db = await getDatabase() + const current = await db + .selectFrom('storage_locations') + .where('id', '=', location.id) + .select('mergedAt') + .executeTakeFirstOrThrow() + expect(current.mergedAt).not.toBeNull() + const merged = await mergedBytes(storage, location.folderName) + expect(merged.equals(Buffer.concat(parts))).toBe(true) + } finally { + await storage.adapter.deleteFolder(location.folderName) + } + }) + + test.skipIf(process.env.VITEST_STORAGE_DRIVER !== 's3')( + 'composes parts server-side when they satisfy the multipart limits', + { timeout: 60_000 }, + async () => { + const storage = await Storage.fromEnv() + const parts = [Buffer.alloc(5 * 1024 * 1024, 'a'), Buffer.alloc(1024, 'b')] + const uploadStream = vi.spyOn(storage.adapter, 'uploadStream') + + const location = await uploadParts(storage, parts) + try { + uploadStream.mockClear() + await storage.waitForOngoingMerges() + expect(uploadStream).not.toHaveBeenCalled() + const merged = await mergedBytes(storage, location.folderName) + expect(merged.equals(Buffer.concat(parts))).toBe(true) + } finally { + uploadStream.mockRestore() + await storage.adapter.deleteFolder(location.folderName) + } + }, + ) +}) diff --git a/tests/setup.ts b/tests/setup.ts index 18c1712..aa5af4e 100644 --- a/tests/setup.ts +++ b/tests/setup.ts @@ -41,6 +41,7 @@ const TESTING_ENV_BASE = { | 'CACHE_FILESYSTEM_MAX_USAGE_PERCENT' | 'ORPHANED_STORAGE_GRACE_PERIOD_HOURS' | 'ENABLE_DIRECT_DOWNLOADS' + | 'EAGER_MERGE' | 'BENCHMARK' | 'SKIP_TOKEN_VALIDATION' | 'ACTIONS_TOKEN_ISSUER' diff --git a/tests/storage-lifecycle.test.ts b/tests/storage-lifecycle.test.ts index a578c0d..7201cc0 100644 --- a/tests/storage-lifecycle.test.ts +++ b/tests/storage-lifecycle.test.ts @@ -53,8 +53,8 @@ describe('storage lifecycle reconciliation', () => { async countFilesInFolder() { return 0 }, - async getFolderSize() { - return 0 + async listFolder() { + return [] }, async clear() {}, } satisfies StorageAdapter @@ -84,8 +84,8 @@ describe('storage lifecycle reconciliation', () => { async countFilesInFolder() { return 0 }, - async getFolderSize() { - return 0 + async listFolder() { + return [] }, async clear() {}, } satisfies StorageAdapter From 67fc7d2122775278740d045804211495f46c7247 Mon Sep 17 00:00:00 2001 From: Louis Haftmann <30736553+LouisHaftmann@users.noreply.github.com> Date: Sun, 13 Sep 2026 13:48:45 +0200 Subject: [PATCH 2/3] feat(storage): opt-in eager merge with server-side S3 compose With EAGER_MERGE=true the merge starts right after upload completion instead of on the first download, so the first restore can be a direct download. On S3 the merge runs inside the bucket via UploadPartCopy when every part but the last is at least 5 MiB; otherwise the existing streaming merge runs immediately. Lazy and eager merges share one lease-fenced runner. Upload completion lists the parts folder once to derive part count, size and per-part sizes. Refs #262 Created with AI. Verified by a human. Co-Authored-By: Claude Fable 5.1 --- CONTEXT.md | 2 +- lib/storage.ts | 51 +++++++++++++++------------------------ tests/eager-merge.test.ts | 21 +++++++++++++--- 3 files changed, 38 insertions(+), 36 deletions(-) diff --git a/CONTEXT.md b/CONTEXT.md index ca18f3e..761c24c 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -49,7 +49,7 @@ _Avoid_: pre-merge, upfront merge **Server-side Merge**: A Merge the storage backend performs by copying Parts into the merged object without their bytes passing through the server (S3 `UploadPartCopy`). Requires every Part to satisfy the backend's limits. -_Avoid_: server-side copy, compose +_Avoid_: server-side copy **Merge Lease**: A time-bound, fenced claim granting one worker authority to complete a Merge. diff --git a/lib/storage.ts b/lib/storage.ts index c75e937..c444e1d 100644 --- a/lib/storage.ts +++ b/lib/storage.ts @@ -29,7 +29,7 @@ import { getSignedUrl } from '@aws-sdk/s3-request-presigner' import { Storage as GcsClient } from '@google-cloud/storage' import { NodeHttpHandler } from '@smithy/node-http-handler' import { sql } from 'kysely' -import { chunk, range } from 'remeda' +import { chunk } from 'remeda' import { match } from 'ts-pattern' import { getDatabase, retryOnLockConflict } from './db' import { env } from './env' @@ -226,14 +226,14 @@ export class Storage { /** * Runs a Merge under a Merge Lease. `write` produces `${folderName}/merged`; - * a lease-fenced transaction then marks the Merge complete. Returns nothing + * a lease-fenced transaction then marks the Merge complete. Returns false * when another worker holds the lease. Writing straight to the final object * is safe (ADR-0004). The returned promise never rejects: failures are logged * and the merge state rolled back. */ private async startMerge(location: StorageLocation, write: () => Promise) { const mergeToken = await acquireMergeLease(this.db, location.id) - if (!mergeToken) return + if (!mergeToken) return false await this.db .updateTable('storage_locations') @@ -299,14 +299,13 @@ export class Storage { * these Parts, otherwise the streaming merge right away. Only the lease * acquisition is awaited; the Merge itself runs in the background. */ - private async mergeEagerly(location: StorageLocation, partSizes: (number | undefined)[]) { + private async mergeEagerly(location: StorageLocation, partSizes: number[]) { const compose = this.adapter.composeParts const composable = compose && partSizes.length <= compose.maxParts && partSizes.every( (bytes, index) => - bytes !== undefined && bytes <= compose.maxPartBytes && (index === partSizes.length - 1 || bytes >= compose.minPartBytes), ) @@ -406,8 +405,9 @@ export class Storage { ) } - const sizeByIndex = new Map(parts.map(({ name, bytes }) => [Number(name), bytes])) - const partSizes = Array.from({ length: partCount }, (_, index) => sizeByIndex.get(index)) + const partSizes = parts + .toSorted((a, b) => Number(a.name) - Number(b.name)) + .map(({ bytes }) => bytes) const location: StorageLocation = { id: randomUUID(), folderName: upload.folderName, @@ -418,7 +418,6 @@ export class Storage { lastDownloadedAt: null, sizeBytes: parts.reduce((sum, { bytes }) => sum + bytes, 0), } - const locationId = location.id await this.db.transaction().execute(async (tx) => { await tx.insertInto('storage_locations').values(location).execute() @@ -438,7 +437,7 @@ export class Storage { .updateTable('cache_entries') .set({ updatedAt: Date.now(), - locationId, + locationId: location.id, }) .where('id', '=', existingCacheEntry.id) .execute() @@ -450,7 +449,7 @@ export class Storage { version: upload.version, id: randomUUID(), updatedAt: Date.now(), - locationId, + locationId: location.id, scope, repoId, }) @@ -798,7 +797,7 @@ export interface StorageAdapter { objectExists(objectName: string): Promise deleteFolder(folderName: string): Promise countFilesInFolder(folderName: string): Promise - /** Direct children of a folder, names relative to it. */ + /** Objects under a folder, names relative to it. */ listFolder(folderName: string): Promise listStorageFolders(): Promise createDownloadUrl?(objectName: string, expiresAt: number): Promise @@ -897,27 +896,17 @@ class S3Adapter implements StorageAdapter { if (!UploadId) throw new Error('S3 did not return an UploadId') try { const Parts = [] - const batches = chunk(range(0, partCount), 8) - for (const indexes of batches) { - const copies = await Promise.all( - indexes.map((index) => - this.s3.send( - new UploadPartCopyCommand({ - Bucket, - Key, - UploadId, - PartNumber: index + 1, - CopySource: `${Bucket}/${this.keyPrefix}/${folderName}/parts/${index}`, - }), - ), - ), - ) - Parts.push( - ...copies.map((copy, offset) => ({ - PartNumber: indexes[offset]! + 1, - ETag: copy.CopyPartResult?.ETag, - })), + for (let PartNumber = 1; PartNumber <= partCount; PartNumber++) { + const copy = await this.s3.send( + new UploadPartCopyCommand({ + Bucket, + Key, + UploadId, + PartNumber, + CopySource: `${Bucket}/${this.keyPrefix}/${folderName}/parts/${PartNumber - 1}`, + }), ) + Parts.push({ PartNumber, ETag: copy.CopyPartResult?.ETag }) } await this.s3.send( new CompleteMultipartUploadCommand({ Bucket, Key, UploadId, MultipartUpload: { Parts } }), diff --git a/tests/eager-merge.test.ts b/tests/eager-merge.test.ts index deb2e81..48005dd 100644 --- a/tests/eager-merge.test.ts +++ b/tests/eager-merge.test.ts @@ -63,23 +63,36 @@ describe('eager merge', () => { } }) + test('leaves the merge to the first download when disabled', async () => { + env.EAGER_MERGE = false + const storage = await Storage.fromEnv() + + const location = await uploadParts(storage, [Buffer.alloc(1024, 'a')]) + try { + await storage.waitForOngoingMerges() + expect(location.mergedAt).toBeNull() + expect(await storage.adapter.objectExists(`${location.folderName}/merged`)).toBe(false) + } finally { + await storage.adapter.deleteFolder(location.folderName) + } + }) + test.skipIf(process.env.VITEST_STORAGE_DRIVER !== 's3')( 'composes parts server-side when they satisfy the multipart limits', { timeout: 60_000 }, async () => { const storage = await Storage.fromEnv() const parts = [Buffer.alloc(5 * 1024 * 1024, 'a'), Buffer.alloc(1024, 'b')] - const uploadStream = vi.spyOn(storage.adapter, 'uploadStream') + const run = vi.spyOn(storage.adapter.composeParts!, 'run') const location = await uploadParts(storage, parts) try { - uploadStream.mockClear() await storage.waitForOngoingMerges() - expect(uploadStream).not.toHaveBeenCalled() + expect(run).toHaveBeenCalledOnce() const merged = await mergedBytes(storage, location.folderName) expect(merged.equals(Buffer.concat(parts))).toBe(true) } finally { - uploadStream.mockRestore() + run.mockRestore() await storage.adapter.deleteFolder(location.folderName) } }, From dc214bf358a2fa159d22cbd4d8f496d290154194 Mon Sep 17 00:00:00 2001 From: Louis Haftmann <30736553+LouisHaftmann@users.noreply.github.com> Date: Sun, 13 Sep 2026 15:07:49 +0200 Subject: [PATCH 3/3] feat(storage): server-side merge on GCS via compose Folds up to 32 sources per call into a top-level temp object and composes `merged` in one final call. No minimum part size, so GCS never falls back to streaming for size reasons. Created with AI. Verified by a human. Co-Authored-By: Claude Fable 5.1 --- ...composes-server-side-on-object-storage.md} | 6 ++-- .../github-actions-cache-server/values.yaml | 6 ++-- lib/storage.ts | 29 +++++++++++++++++++ tests/eager-merge.test.ts | 24 +++++++++++++-- 4 files changed, 57 insertions(+), 8 deletions(-) rename docs/adr/{0009-eager-merge-is-opt-in-and-composes-server-side-on-s3.md => 0009-eager-merge-is-opt-in-and-composes-server-side-on-object-storage.md} (50%) diff --git a/docs/adr/0009-eager-merge-is-opt-in-and-composes-server-side-on-s3.md b/docs/adr/0009-eager-merge-is-opt-in-and-composes-server-side-on-object-storage.md similarity index 50% rename from docs/adr/0009-eager-merge-is-opt-in-and-composes-server-side-on-s3.md rename to docs/adr/0009-eager-merge-is-opt-in-and-composes-server-side-on-object-storage.md index df0895c..ff8faed 100644 --- a/docs/adr/0009-eager-merge-is-opt-in-and-composes-server-side-on-s3.md +++ b/docs/adr/0009-eager-merge-is-opt-in-and-composes-server-side-on-object-storage.md @@ -1,9 +1,9 @@ -# Eager Merge is opt-in and composes server-side on S3 +# Eager Merge is opt-in and composes server-side on object storage Direct downloads need a merged object, and the Merge runs lazily on first download, so the first Cache Hit of every entry is proxied through the server even with `ENABLE_DIRECT_DOWNLOADS`. `EAGER_MERGE=true` runs the Merge right after upload completion instead. It stays off by default because lazy merging never pays merge traffic for entries that are never downloaded. -Both paths share one merge runner. Merge Lease, renewal, lease-fenced completion, and rollback are identical, so eager and lazy merges race safely against each other and against `cleanup:merges`. Upload completion picks the strategy up front from the Part sizes it already lists to record `sizeBytes`. When the adapter offers `composeParts` and every Part satisfies its limits (on S3: every Part but the last at least 5 MiB, none above 5 GiB, at most 10,000 Parts), the Merge is a Server-side Merge: `CreateMultipartUpload`, one `UploadPartCopy` per Part, `CompleteMultipartUpload`, with no bytes passing through the server. Otherwise the existing streaming merge runs immediately. Buildx uploads 1 MiB blocks, so its entries always take the streaming path. A backend that rejects `UploadPartCopy` fails the Merge like any other merge failure; the entry then merges lazily on first download. +Both paths share one merge runner. Merge Lease, renewal, lease-fenced completion, and rollback are identical, so eager and lazy merges race safely against each other and against `cleanup:merges`. Upload completion picks the strategy up front from the Part sizes it already lists to record `sizeBytes`. When the adapter offers `composeParts` and every Part satisfies its limits, the Merge is a Server-side Merge with no bytes passing through the server. On S3 that is `CreateMultipartUpload`, one `UploadPartCopy` per Part, `CompleteMultipartUpload`; the limits are every Part but the last at least 5 MiB, none above 5 GiB, at most 10,000 Parts. On GCS it is `compose`, folding 32 sources at a time into a top-level temp object and composing `merged` in one final call; there is no minimum Part size. Otherwise the existing streaming merge runs immediately. Buildx uploads 1 MiB blocks, so its entries always take the streaming path. A backend that rejects the copy call fails the Merge like any other merge failure; the entry then merges lazily on first download. This does not reopen ADR-0004. The 5 GiB cap it cites applies to `CopyObject` used as a promote step. `UploadPartCopy` writes `merged` directly under the same lease fence, and Parts are immutable, so a merger that lost its lease can only write identical bytes. -Costs: every entry occupies twice its size until `cleanup:parts` removes the Parts, not only entries that were downloaded, and `sizeBytes` still counts Parts only. A worker killed mid-composition leaves an incomplete multipart upload in the bucket; configure an `AbortIncompleteMultipartUpload` lifecycle rule. `FinalizeCacheEntryUpload` waits only for the Merge Lease, not for the Merge. GCS `compose` is not implemented; GCS entries take the streaming path. +Costs: every entry occupies twice its size until `cleanup:parts` removes the Parts, not only entries that were downloaded, and `sizeBytes` still counts Parts only. A worker killed mid-composition leaves an incomplete multipart upload on S3 (configure an `AbortIncompleteMultipartUpload` lifecycle rule) or a temp object on GCS (reclaimed as Orphaned Storage). `FinalizeCacheEntryUpload` waits only for the Merge Lease, not for the Merge. The filesystem driver has no server-side copy and always streams. diff --git a/install/kubernetes/github-actions-cache-server/values.yaml b/install/kubernetes/github-actions-cache-server/values.yaml index c544a7a..47bfaa1 100644 --- a/install/kubernetes/github-actions-cache-server/values.yaml +++ b/install/kubernetes/github-actions-cache-server/values.yaml @@ -40,9 +40,9 @@ config: enableDirectDownloads: false # -- Merge cache parts right after upload instead of on first download, so - # the first restore can be a direct download. On S3 the merge happens inside - # the bucket (UploadPartCopy) when every part but the last is at least 5 MiB. - # Doubles storage per entry until the parts cleanup job runs. + # the first restore can be a direct download. On S3 (UploadPartCopy, parts + # but the last at least 5 MiB) and GCS (compose) the merge happens inside the + # bucket. Doubles storage per entry until the parts cleanup job runs. eagerMerge: false # -- Number of days to keep stale cache data before deleting it. Set to 0 to disable. diff --git a/lib/storage.ts b/lib/storage.ts index c444e1d..b33bd5d 100644 --- a/lib/storage.ts +++ b/lib/storage.ts @@ -1281,6 +1281,35 @@ class GcsAdapter implements StorageAdapter { private bucket private keyPrefix = 'gh-actions-cache' + // GCS compose takes at most 32 sources per call and has no minimum size: + // https://cloud.google.com/storage/docs/composing-objects + composeParts = { + minPartBytes: 0, + maxPartBytes: 5 * 1024 ** 4, + maxParts: Infinity, + run: async (folderName: string, partCount: number) => { + const sources = Array.from({ length: partCount }, (_, index) => + this.bucket.file(`${this.keyPrefix}/${folderName}/parts/${index}`), + ) + // Folds batches into a top-level temp object (reclaimed as Orphaned + // Storage if we crash) so `merged` only ever appears in one final, + // atomic compose (ADR-0004). + const temp = this.bucket.file(`${this.keyPrefix}/tmp-${randomUUID()}`) + try { + while (sources.length > 32) { + await this.bucket.combine(sources.splice(0, 32), temp) + sources.unshift(temp) + } + await this.bucket.combine( + sources, + this.bucket.file(`${this.keyPrefix}/${folderName}/merged`), + ) + } finally { + if (partCount > 32) await temp.delete({ ignoreNotFound: true }) + } + }, + } + constructor({ bucket, gcs }: { bucket: string; gcs: GcsClient }) { this.bucket = gcs.bucket(bucket) } diff --git a/tests/eager-merge.test.ts b/tests/eager-merge.test.ts index 48005dd..5f37bb0 100644 --- a/tests/eager-merge.test.ts +++ b/tests/eager-merge.test.ts @@ -77,8 +77,8 @@ describe('eager merge', () => { } }) - test.skipIf(process.env.VITEST_STORAGE_DRIVER !== 's3')( - 'composes parts server-side when they satisfy the multipart limits', + test.skipIf((process.env.VITEST_STORAGE_DRIVER ?? 'filesystem') === 'filesystem')( + 'composes parts server-side when they satisfy the backend limits', { timeout: 60_000 }, async () => { const storage = await Storage.fromEnv() @@ -97,4 +97,24 @@ describe('eager merge', () => { } }, ) + + test.skipIf(process.env.VITEST_STORAGE_DRIVER !== 'gcs')( + 'folds more than 32 parts through a temp object and removes it', + { timeout: 60_000 }, + async () => { + const storage = await Storage.fromEnv() + const parts = Array.from({ length: 33 }, (_, index) => Buffer.alloc(1024, String(index % 10))) + + const location = await uploadParts(storage, parts) + try { + await storage.waitForOngoingMerges() + const merged = await mergedBytes(storage, location.folderName) + expect(merged.equals(Buffer.concat(parts))).toBe(true) + const folders = await storage.adapter.listStorageFolders() + expect(folders.filter(({ folderName }) => folderName.startsWith('tmp-'))).toEqual([]) + } finally { + await storage.adapter.deleteFolder(location.folderName) + } + }, + ) })