From e4caccdeabca92b56f08f6034fb5a3c3a400fb8e Mon Sep 17 00:00:00 2001 From: kptdobe Date: Tue, 26 May 2026 10:57:16 +0200 Subject: [PATCH] fix(audit): append-only ledger eliminates writeAuditEntry contention Replace audit.txt read-modify-write-with-If-Match + 6-retry loop with one unconditional PUT per entry under {org}/{repo}/.da-versions/{fileId}/audit/{ts}-{rand}.txt. COR-39's retry-budget bump (#282, 4->6 retries / ~3050 ms) did not converge (Day 3 post-deploy: 6814 writeAuditEntry-failed / 24h vs 5614 baseline); the contention window outlasts the retry budget under burst. Removing the shared key removes the contention by construction. Read path already merges everything under the audit prefix via ListObjectsV2, so legacy audit.txt + audit-{ts}.txt archives + new per-entry objects are transparently consumed. Same-user same-window collapse moves from write-side to read-side post-processing (AUDIT_TIME_WINDOW_MS unchanged) so the observable audit sequence is preserved. - src/storage/version/audit.js: writeAuditEntry single unconditional PUT, readAuditLines applies the collapse pass, paginates ListObjectsV2. - src/storage/version/paths.js: add auditEntryKey(repo, fileId, ts, rand). - test/storage/version/audit.test.js: replace write-side collapse/retry/archive tests with read-side collapse + per-entry PUT contract tests; keep mixed-file read tests to prove transparent migration. - test/storage/object/conditionals.test.js: PUT count drops from 8 to 2 and the elevated 8000 ms mocha timeout is no longer needed. - test/storage/version/paths.test.js: cover auditEntryKey. - test/storage/version/put.test.js: comment update only. Acceptance: writeAuditEntry-failed < 50/24h sustained for 3 consecutive daily reviews post-deploy. Zero PreconditionFailed on the audit-write path by construction. Compaction deferred until measured (existing archive cycle still applies; per-entry objects throttle naturally via ListObjectsV2 page size). Co-Authored-By: Paperclip --- src/storage/version/audit.js | 196 ++-- src/storage/version/paths.js | 16 +- test/storage/object/conditionals.test.js | 9 +- test/storage/version/audit.test.js | 1207 +++++++--------------- test/storage/version/paths.test.js | 24 +- test/storage/version/put.test.js | 4 +- 6 files changed, 487 insertions(+), 969 deletions(-) diff --git a/src/storage/version/audit.js b/src/storage/version/audit.js index c8964249..0a20c92b 100644 --- a/src/storage/version/audit.js +++ b/src/storage/version/audit.js @@ -17,13 +17,11 @@ import { } from '@aws-sdk/client-s3'; import getS3Config from '../utils/config.js'; -import { auditKey, auditArchiveKey, auditDirPrefix } from './paths.js'; +import { auditDirPrefix, auditEntryKey } from './paths.js'; /** Same-user edits within this window (ms) collapse into one entry (last timestamp). 30 min. */ export const AUDIT_TIME_WINDOW_MS = 30 * 60 * 1000; -export const AUDIT_MAX_ENTRIES = 500; - const SEP = '\t'; /** @@ -69,9 +67,9 @@ export function parseAuditLine(line) { } /** - * Read audit.txt body stream to string. Handles Web ReadableStream (Workers/R2), + * Read audit body stream to string. Handles Web ReadableStream (Workers/R2), * fetch Response.body, and Node-style async iterable streams. - * @param {ReadableStream|import('stream').Readable|string} body + * @param {ReadableStream|import("stream").Readable|string} body * @returns {Promise} */ async function streamToString(body) { @@ -118,25 +116,76 @@ async function streamToString(body) { } /** - * Read all audit lines for a file (new structure). + * Normalize users for same-user comparison (stable string). + * @param {string} usersJson + * @returns {string} + */ +function usersNormalized(usersJson) { + try { + const arr = JSON.parse(usersJson); + const emails = Array.isArray(arr) ? arr.map((u) => u?.email ?? '').filter(Boolean) : []; + return emails.join(',') || usersJson; + } catch { + return usersJson; + } +} + +/** + * Collapse consecutive same-user edit entries within AUDIT_TIME_WINDOW_MS into the later entry. + * Matches the historical write-side collapse: same user, both edits (no versionLabel/versionId), + * within 30 minutes => keep only the later entry. Version entries always break the window. + * @param {object[]} entries - ascending by timestamp; each carries usersRaw for collapse + * @returns {object[]} + */ +function collapseAuditEntries(entries) { + const out = []; + for (const entry of entries) { + const last = out.length ? out[out.length - 1] : null; + const isVersion = !!(entry.versionLabel || entry.versionId); + const lastIsVersion = last && !!(last.versionLabel || last.versionId); + const canCollapse = last + && !isVersion + && !lastIsVersion + && usersNormalized(last.usersRaw) === usersNormalized(entry.usersRaw) + && (entry.timestamp - last.timestamp) <= AUDIT_TIME_WINDOW_MS; + if (canCollapse) { + out[out.length - 1] = entry; + } else { + out.push(entry); + } + } + return out; +} + +/** + * Read all audit entries for a file. Merges legacy audit.txt + audit-{ts}.txt archives + per-entry + * objects under {org}/{repo}/.da-versions/{fileId}/audit/. Entries are sorted ascending by + * timestamp with the same-user same-window collapse applied (formerly enforced write-side). * @param {object} env - * @param {{ bucket: string, org: string }} ctx - bucket, org + * @param {{ bucket: string, org: string }} ctx * @param {string} repo * @param {string} fileId - * @returns {Promise<{ timestamp: number, users: object[], path: string }[]>} + * @returns {Promise} entries with { timestamp, users, path, versionLabel?, versionId? } */ export async function readAuditLines(env, ctx, repo, fileId) { const config = getS3Config(env); const client = new S3Client(config); const prefix = `${ctx.org}/${auditDirPrefix(repo, fileId)}`; - let keys; + let keys = []; try { - const listResp = await client.send(new ListObjectsV2Command({ - Bucket: ctx.bucket, - Prefix: prefix, - })); - keys = (listResp.Contents || []).map((obj) => obj.Key); + let continuationToken; + do { + // eslint-disable-next-line no-await-in-loop + const listResp = await client.send(new ListObjectsV2Command({ + Bucket: ctx.bucket, + Prefix: prefix, + ContinuationToken: continuationToken, + })); + const page = (listResp.Contents || []).map((obj) => obj.Key); + keys = keys.concat(page); + continuationToken = listResp.IsTruncated ? listResp.NextContinuationToken : undefined; + } while (continuationToken); } catch (e) { if (e.$metadata?.httpStatusCode === 404 || e.name === 'NoSuchKey') { return []; @@ -156,8 +205,9 @@ export async function readAuditLines(env, ctx, repo, fileId) { } })); - return allLineArrays.flat().map((line) => ({ + const parsed = allLineArrays.flat().map((line) => ({ timestamp: parseInt(line.timestamp, 10) || 0, + usersRaw: line.users, users: (() => { try { return JSON.parse(line.users); @@ -169,122 +219,38 @@ export async function readAuditLines(env, ctx, repo, fileId) { versionLabel: line.versionLabel || undefined, versionId: line.versionId || undefined, })); -} -/** - * Normalize users for same-user comparison (stable string). - * @param {string} usersJson - * @returns {string} - */ -function usersNormalized(usersJson) { - try { - const arr = JSON.parse(usersJson); - const emails = Array.isArray(arr) ? arr.map((u) => u?.email ?? '').filter(Boolean) : []; - return emails.join(',') || usersJson; - } catch { - return usersJson; - } + parsed.sort((a, b) => a.timestamp - b.timestamp); + + return collapseAuditEntries(parsed).map(({ usersRaw: _, ...rest }) => rest); } /** - * Append or update last line in audit.txt (read-modify-write). If last line is same user - * and within AUDIT_TIME_WINDOW_MS and both last and new are edits (no version), replace that - * line; else append. A version entry always appends and is never replaced (breaks the window). - * - * Uses If-Match on the PUT so that a concurrent write causes a 412, which triggers up to 6 - * retries with random jitter to reduce thundering-herd contention (7 total attempts). + * Append one audit entry as a fresh per-entry object under the audit/ prefix. Append-only: + * no GET, no etag, no retry — one unconditional PUT per call eliminates the read-modify-write + * contention that the previous read-modify-write-with-If-Match path exhibited under load. * @param {object} env * @param {{ bucket: string, org: string }} ctx - bucket, org * @param {string} repo * @param {string} fileId * @param {object} entry - { timestamp, users, path, versionLabel?, versionId? } - * @param {number} [attempt=0] - retry counter (max 6 retries) - * @returns {Promise<{ status: number }>} + * @returns {Promise<{ status: number, error?: string }>} */ -export async function writeAuditEntry(env, ctx, repo, fileId, entry, attempt = 0) { +export async function writeAuditEntry(env, ctx, repo, fileId, entry) { try { const config = getS3Config(env); const client = new S3Client(config); - const key = `${ctx.org}/${auditKey(repo, fileId)}`; - const nowMs = parseInt(entry.timestamp, 10) || Date.now(); - const entryUsersNorm = usersNormalized(entry.users); - - let existingText = ''; - let etag; - try { - const getResp = await client.send(new GetObjectCommand({ - Bucket: ctx.bucket, - Key: key, - })); - const body = getResp?.Body; - existingText = body != null ? await streamToString(body) : ''; - etag = getResp?.ETag; - } catch (e) { - if (e?.$metadata?.httpStatusCode !== 404 && e?.name !== 'NoSuchKey') { - throw e; - } - } - - const lines = existingText.split('\n').filter((l) => l.trim()); - const lastLine = lines.length ? parseAuditLine(lines[lines.length - 1]) : null; - - const isVersionEntry = (entry.versionLabel ?? '') !== '' || (entry.versionId ?? '') !== ''; - const lastIsVersion = lastLine - && ((lastLine.versionLabel ?? '') !== '' || (lastLine.versionId ?? '') !== ''); - const canCollapse = lastLine - && usersNormalized(lastLine.users) === entryUsersNorm - && !isVersionEntry - && !lastIsVersion - && (nowMs - (parseInt(lastLine.timestamp, 10) || 0) <= AUDIT_TIME_WINDOW_MS); - let newContent; - if (canCollapse) { - lines[lines.length - 1] = formatAuditLine(entry); - newContent = `${lines.join('\n')}\n`; - } else { - const sep = existingText && !existingText.endsWith('\n') ? '\n' : ''; - newContent = `${existingText}${sep}${formatAuditLine(entry)}\n`; - } - - const shouldArchive = !canCollapse && lines.length >= AUDIT_MAX_ENTRIES; - if (shouldArchive) { - const archiveTs = lastLine?.timestamp || Date.now(); - await client.send(new PutObjectCommand({ - Bucket: ctx.bucket, - Key: `${ctx.org}/${auditArchiveKey(repo, fileId, archiveTs)}`, - Body: existingText, - ContentType: 'text/plain; charset=utf-8', - })); - newContent = `${formatAuditLine(entry)}\n`; - } - - const putInput = { + const ts = parseInt(entry.timestamp, 10) || Date.now(); + const rand = crypto.randomUUID().replace(/-/g, '').slice(0, 16); + const key = `${ctx.org}/${auditEntryKey(repo, fileId, ts, rand)}`; + const body = `${formatAuditLine({ ...entry, timestamp: String(ts) })}\n`; + const resp = await client.send(new PutObjectCommand({ Bucket: ctx.bucket, Key: key, - Body: newContent, + Body: body, ContentType: 'text/plain; charset=utf-8', - }; - if (etag) putInput.IfMatch = etag; - - try { - const resp = await client.send(new PutObjectCommand(putInput)); - return { status: resp?.$metadata?.httpStatusCode ?? 200 }; - } catch (e) { - if (e?.$metadata?.httpStatusCode === 412 && attempt < 6) { - // Exponential jitter, 6 retries: per-attempt max 50, 100, 200, - // 400, 800, 1600 ms (~3050 ms worst-case total). Two prior 4-retry - // backoff bumps did not converge because the contention window is - // wider than per-write latency — every retry inside a short window - // observes the same losing-etag generation. Append-only ledger - // (one object per entry) is the next escalation if this still fails. - const delay = Math.random() * 50 * 2 ** attempt; - // eslint-disable-next-line no-await-in-loop -- sequential retry with jitter - await new Promise((r) => { - setTimeout(r, delay); - }); - return writeAuditEntry(env, ctx, repo, fileId, entry, attempt + 1); - } - throw e; - } + })); + return { status: resp?.$metadata?.httpStatusCode ?? 200 }; } catch (e) { // eslint-disable-next-line no-console console.error('writeAuditEntry failed', e); diff --git a/src/storage/version/paths.js b/src/storage/version/paths.js index 722b5d22..edea192a 100644 --- a/src/storage/version/paths.js +++ b/src/storage/version/paths.js @@ -44,7 +44,7 @@ export function auditArchiveKey(repo, fileId, timestamp) { } /** - * Prefix that matches all audit files (audit.txt + audit-*.txt) for a file. + * Prefix matching all audit files (audit.txt + audit-*.txt + per-entry objects under audit/). * @param {string} repo * @param {string} fileId * @returns {string} prefix (repo/.da-versions/fileId/audit) @@ -52,3 +52,17 @@ export function auditArchiveKey(repo, fileId, timestamp) { export function auditDirPrefix(repo, fileId) { return `${repo}/.da-versions/${fileId}/audit`; } + +/** + * Per-entry audit object key (append-only ledger). Each audit entry is one S3 object + * under {repo}/.da-versions/{fileId}/audit/, eliminating read-modify-write contention on + * audit.txt. Uses {ts}-{rand} so concurrent writers at the same millisecond cannot collide. + * @param {string} repo + * @param {string} fileId + * @param {string|number} timestamp - entry timestamp (ms) + * @param {string} rand - random suffix for uniqueness + * @returns {string} key (repo/.da-versions/fileId/audit/{ts}-{rand}.txt) + */ +export function auditEntryKey(repo, fileId, timestamp, rand) { + return `${repo}/.da-versions/${fileId}/audit/${timestamp}-${rand}.txt`; +} diff --git a/test/storage/object/conditionals.test.js b/test/storage/object/conditionals.test.js index fe4310c8..48c480aa 100644 --- a/test/storage/object/conditionals.test.js +++ b/test/storage/object/conditionals.test.js @@ -228,10 +228,7 @@ describe('Conditional Headers', () => { assert.strictEqual(resp.status, 200); }); - it('returns 412 when ETag does not match and does not retry', async function returnsImmediateOn412NoRetry() { - // writeAuditEntry's 6-retry exponential jitter has a worst-case budget of - // ~3050 ms, which exceeds mocha's 2000 ms default; bound this test above it. - this.timeout(8000); + it('returns 412 when ETag does not match and does not retry', async () => { const existingEtag = '"existing123"'; s3Mock .on(GetObjectCommand) @@ -265,8 +262,8 @@ describe('Conditional Headers', () => { // Should return 412 and NOT retry the main PUT assert.strictEqual(resp.status, 412); - // 7 audit PUT attempts (1 initial + 6 retries on 412) + 1 main PUT = 8 total - assert.strictEqual(s3Mock.commandCalls(PutObjectCommand).length, 8); + // Append-only audit ledger: 1 main PUT (returns 412) + 1 unconditional audit PUT = 2 total + assert.strictEqual(s3Mock.commandCalls(PutObjectCommand).length, 2); }); }); diff --git a/test/storage/version/audit.test.js b/test/storage/version/audit.test.js index f8d04ca0..15161c52 100644 --- a/test/storage/version/audit.test.js +++ b/test/storage/version/audit.test.js @@ -13,10 +13,34 @@ import assert from 'node:assert'; import esmock from 'esmock'; import { GetObjectCommand, PutObjectCommand, ListObjectsV2Command } from '@aws-sdk/client-s3'; +const AUDIT_MODULE = '../../../src/storage/version/audit.js'; +const CONFIG_MODULE = '../../../src/storage/utils/config.js'; + +function makeStreamBody(text) { + return new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(text)); + controller.close(); + }, + }); +} + +async function mockAudit(sendHandler) { + return esmock(AUDIT_MODULE, { + '@aws-sdk/client-s3': { + S3Client: function S3Client() { this.send = sendHandler; }, + GetObjectCommand, + PutObjectCommand, + ListObjectsV2Command, + }, + [CONFIG_MODULE]: { default: () => ({}) }, + }); +} + describe('Version Audit', () => { describe('formatAuditLine / parseAuditLine', () => { it('round-trips one entry (edit, no versionLabel/versionId)', async () => { - const { formatAuditLine, parseAuditLine } = await import('../../../src/storage/version/audit.js'); + const { formatAuditLine, parseAuditLine } = await import(AUDIT_MODULE); const entry = { timestamp: '1000', users: '[{"email":"a@b.com"}]', path: 'repo/doc.html' }; const line = formatAuditLine(entry); assert.strictEqual(line, '1000\t[{"email":"a@b.com"}]\trepo/doc.html\t\t'); @@ -29,7 +53,7 @@ describe('Version Audit', () => { }); it('round-trips entry with versionLabel and versionId (labelled save)', async () => { - const { formatAuditLine, parseAuditLine } = await import('../../../src/storage/version/audit.js'); + const { formatAuditLine, parseAuditLine } = await import(AUDIT_MODULE); const entry = { timestamp: '2000', users: '[{"email":"u@x.com"}]', @@ -45,56 +69,36 @@ describe('Version Audit', () => { }); it('parses legacy 3-column line (backward compat)', async () => { - const { parseAuditLine } = await import('../../../src/storage/version/audit.js'); - const line = '1000\t[{}]\trepo/f.html'; - const parsed = parseAuditLine(line); + const { parseAuditLine } = await import(AUDIT_MODULE); + const parsed = parseAuditLine('1000\t[{}]\trepo/f.html'); assert.strictEqual(parsed.timestamp, '1000'); assert.strictEqual(parsed.versionLabel, ''); assert.strictEqual(parsed.versionId, ''); }); it('parses legacy 4-column line (versionId only, no label)', async () => { - const { parseAuditLine } = await import('../../../src/storage/version/audit.js'); - const line = '2000\t[{}]\trepo/f.html\told-uuid.html'; - const parsed = parseAuditLine(line); + const { parseAuditLine } = await import(AUDIT_MODULE); + const parsed = parseAuditLine('2000\t[{}]\trepo/f.html\told-uuid.html'); assert.strictEqual(parsed.timestamp, '2000'); assert.strictEqual(parsed.versionLabel, ''); assert.strictEqual(parsed.versionId, 'old-uuid.html'); }); }); - describe('readAuditLines', () => { it('reads audit lines from a Node-style async iterable body', async () => { const lineText = '3000\t[{"email":"node@x.com"}]\trepo/path.html\t\t\n'; - async function* asyncIterableBody() { yield Buffer.from(lineText.slice(0, 10)); yield Buffer.from(lineText.slice(10)); } - - const { readAuditLines } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof ListObjectsV2Command) { - return { Contents: [{ Key: 'o/repo/.da-versions/fid/audit.txt' }] }; - } - if (cmd instanceof GetObjectCommand) return { Body: asyncIterableBody() }; - return {}; - }; - }, - GetObjectCommand, - PutObjectCommand, - ListObjectsV2Command, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - + const { readAuditLines } = await mockAudit(async (cmd) => { + if (cmd instanceof ListObjectsV2Command) { + return { Contents: [{ Key: 'o/repo/.da-versions/fid/audit.txt' }] }; + } + if (cmd instanceof GetObjectCommand) return { Body: asyncIterableBody() }; + return {}; + }); const lines = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); - assert.strictEqual(lines.length, 1); assert.strictEqual(lines[0].timestamp, 3000); assert.deepStrictEqual(lines[0].users, [{ email: 'node@x.com' }]); @@ -103,84 +107,44 @@ describe('Version Audit', () => { it('returns [] when S3 throws 404 (NoSuchKey)', async () => { const notFound = Object.assign(new Error('not found'), { name: 'NoSuchKey' }); - - const { readAuditLines } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async () => { - throw notFound; - }; - }, - GetObjectCommand, - PutObjectCommand, - ListObjectsV2Command, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - + const { readAuditLines } = await mockAudit(async () => { + throw notFound; + }); const lines = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); assert.deepStrictEqual(lines, []); }); - it('returns empty string when Node-style async iterable body throws during iteration', async () => { + it('skips a per-entry stream that throws during iteration (no crash)', async () => { + const goodLine = '9000\t[{"email":"good@x.com"}]\trepo/f.html\t\t\n'; async function* throwingIterable() { yield Buffer.from('partial'); throw new Error('stream error mid-read'); } - - const { writeAuditEntry } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof GetObjectCommand) return { Body: throwingIterable() }; - if (cmd instanceof PutObjectCommand) return { $metadata: { httpStatusCode: 200 } }; - return { $metadata: { httpStatusCode: 200 } }; - }; - }, - GetObjectCommand, - PutObjectCommand, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - // throwing iterable → streamToString catch → existingText = '' → append new entry - const result = await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { - timestamp: '7000', - users: '[{"email":"e@x.com"}]', - path: 'repo/f.html', + const { readAuditLines } = await mockAudit(async (cmd) => { + if (cmd instanceof ListObjectsV2Command) { + return { + Contents: [ + { Key: 'o/repo/.da-versions/fid/audit/8000-aaaa.txt' }, + { Key: 'o/repo/.da-versions/fid/audit/9000-bbbb.txt' }, + ], + }; + } + if (cmd instanceof GetObjectCommand) { + if (cmd.input.Key.endsWith('8000-aaaa.txt')) return { Body: throwingIterable() }; + return { Body: makeStreamBody(goodLine) }; + } + return {}; }); - - assert.strictEqual(result.status, 200); + const lines = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); + assert.strictEqual(lines.length, 1, 'failed-stream object must be skipped, not crash readAuditLines'); + assert.strictEqual(lines[0].timestamp, 9000); }); - it('re-throws when S3 throws a non-404 error in readAuditLines', async () => { - const serverError = Object.assign(new Error('server err'), { - $metadata: { httpStatusCode: 500 }, + it('re-throws when S3 throws a non-404 error', async () => { + const serverError = Object.assign(new Error('server err'), { $metadata: { httpStatusCode: 500 } }); + const { readAuditLines } = await mockAudit(async () => { + throw serverError; }); - - const { readAuditLines } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async () => { - throw serverError; - }; - }, - GetObjectCommand, - PutObjectCommand, - ListObjectsV2Command, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - await assert.rejects( () => readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'), (err) => err.$metadata?.httpStatusCode === 500, @@ -189,816 +153,371 @@ describe('Version Audit', () => { it('returns default anonymous user when users JSON is invalid', async () => { const lineText = '1000\tinvalid-json\trepo/doc.html\t\t\n'; - - const { readAuditLines } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof ListObjectsV2Command) { - return { Contents: [{ Key: 'o/repo/.da-versions/fid/audit.txt' }] }; - } - return { - Body: new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode(lineText)); - controller.close(); - }, - }), - }; - }; - }, - GetObjectCommand, - PutObjectCommand, - ListObjectsV2Command, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - + const { readAuditLines } = await mockAudit(async (cmd) => { + if (cmd instanceof ListObjectsV2Command) { + return { Contents: [{ Key: 'o/repo/.da-versions/fid/audit.txt' }] }; + } + return { Body: makeStreamBody(lineText) }; + }); const lines = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); assert.strictEqual(lines.length, 1); assert.deepStrictEqual(lines[0].users, [{ email: 'anonymous' }]); }); - - it('reads current audit.txt and archive files, merging all entries', async () => { + it('merges legacy audit.txt, archive files, and per-entry objects (transparent migration)', async () => { const archiveLine = '1000\t[{"email":"a@x.com"}]\t/doc.html\t\t'; - const currentLine = '9000\t[{"email":"b@x.com"}]\t/doc.html\t\t'; - - const { readAuditLines } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof ListObjectsV2Command) { - return { - Contents: [ - { Key: 'o/repo/.da-versions/fid/audit-1000.txt' }, - { Key: 'o/repo/.da-versions/fid/audit.txt' }, - ], - }; - } - if (cmd instanceof GetObjectCommand) { - const isArchive = cmd.input.Key.includes('audit-1000'); - const line = isArchive ? archiveLine : currentLine; - return { - Body: new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode(`${line}\n`)); - controller.close(); - }, - }), - }; - } - return {}; - }; - }, - GetObjectCommand, - PutObjectCommand, - ListObjectsV2Command, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - const lines = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); - assert.strictEqual(lines.length, 2); - const timestamps = lines.map((l) => l.timestamp).sort((a, b) => a - b); - assert.deepStrictEqual(timestamps, [1000, 9000]); + const legacyLine = '2000\t[{"email":"b@x.com"}]\t/doc.html\t\t'; + const perEntry1 = '7000\t[{"email":"c@x.com"}]\t/doc.html\t\t'; + const perEntry2 = '9000\t[{"email":"d@x.com"}]\t/doc.html\t\t'; + const { readAuditLines } = await mockAudit(async (cmd) => { + if (cmd instanceof ListObjectsV2Command) { + return { + Contents: [ + { Key: 'o/repo/.da-versions/fid/audit.txt' }, + { Key: 'o/repo/.da-versions/fid/audit-1000.txt' }, + { Key: 'o/repo/.da-versions/fid/audit/7000-aaaa.txt' }, + { Key: 'o/repo/.da-versions/fid/audit/9000-bbbb.txt' }, + ], + }; + } + if (cmd instanceof GetObjectCommand) { + const k = cmd.input.Key; + if (k.endsWith('audit-1000.txt')) return { Body: makeStreamBody(`${archiveLine}\n`) }; + if (k.endsWith('audit.txt')) return { Body: makeStreamBody(`${legacyLine}\n`) }; + if (k.endsWith('7000-aaaa.txt')) return { Body: makeStreamBody(`${perEntry1}\n`) }; + if (k.endsWith('9000-bbbb.txt')) return { Body: makeStreamBody(`${perEntry2}\n`) }; + } + return {}; + }); + const out = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); + assert.strictEqual(out.length, 4, 'all four sources must merge transparently'); + assert.deepStrictEqual(out.map((l) => l.timestamp), [1000, 2000, 7000, 9000], 'must be sorted ascending'); }); - it('skips an archive file that throws on GET and still returns other entries', async () => { + it('skips an object that throws on GET and still returns other entries', async () => { const currentLine = '9000\t[{"email":"b@x.com"}]\t/doc.html\t\t'; - - const { readAuditLines } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof ListObjectsV2Command) { - return { - Contents: [ - { Key: 'o/repo/.da-versions/fid/audit-1000.txt' }, - { Key: 'o/repo/.da-versions/fid/audit.txt' }, - ], - }; - } - if (cmd instanceof GetObjectCommand) { - if (cmd.input.Key.includes('audit-1000')) throw new Error('S3 read error'); - return { - Body: new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode(`${currentLine}\n`)); - controller.close(); - }, - }), - }; - } - return {}; - }; - }, - GetObjectCommand, - PutObjectCommand, - ListObjectsV2Command, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - const lines = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); - assert.strictEqual(lines.length, 1, 'failed archive GET must be skipped, not throw'); - assert.strictEqual(lines[0].timestamp, 9000); + const { readAuditLines } = await mockAudit(async (cmd) => { + if (cmd instanceof ListObjectsV2Command) { + return { + Contents: [ + { Key: 'o/repo/.da-versions/fid/audit-1000.txt' }, + { Key: 'o/repo/.da-versions/fid/audit.txt' }, + ], + }; + } + if (cmd instanceof GetObjectCommand) { + if (cmd.input.Key.includes('audit-1000')) throw new Error('S3 read error'); + return { Body: makeStreamBody(`${currentLine}\n`) }; + } + return {}; + }); + const out = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); + assert.strictEqual(out.length, 1, 'failed GET must be skipped, not throw'); + assert.strictEqual(out[0].timestamp, 9000); }); - it('returns [] when no audit files exist (empty list)', async () => { - const { readAuditLines } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof ListObjectsV2Command) return { Contents: [] }; - return {}; - }; - }, - GetObjectCommand, - PutObjectCommand, - ListObjectsV2Command, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - const lines = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); - assert.deepStrictEqual(lines, []); + it('returns [] when no audit objects exist (empty list)', async () => { + const { readAuditLines } = await mockAudit(async (cmd) => { + if (cmd instanceof ListObjectsV2Command) return { Contents: [] }; + return {}; + }); + const out = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); + assert.deepStrictEqual(out, []); }); - }); - describe('writeAuditEntry read-modify-write', () => { - it('appends new line when existing content is read (Web ReadableStream body)', async () => { - const existingLine = '1000\t[{"email":"a@b.com"}]\trepo/path.html'; - const bodyStream = new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode(`${existingLine}\n`)); - controller.close(); - }, + it('follows ContinuationToken to page across more than one ListObjectsV2 result', async () => { + const line1 = '1000\t[{"email":"a@x.com"}]\t/doc.html\t\t'; + const line2 = '2000\t[{"email":"b@x.com"}]\t/doc.html\t\t'; + let listCalls = 0; + const { readAuditLines } = await mockAudit(async (cmd) => { + if (cmd instanceof ListObjectsV2Command) { + listCalls += 1; + if (cmd.input.ContinuationToken === 'tok-1') { + return { Contents: [{ Key: 'o/repo/.da-versions/fid/audit/2000-b.txt' }], IsTruncated: false }; + } + return { + Contents: [{ Key: 'o/repo/.da-versions/fid/audit/1000-a.txt' }], + IsTruncated: true, + NextContinuationToken: 'tok-1', + }; + } + if (cmd instanceof GetObjectCommand) { + const k = cmd.input.Key; + if (k.endsWith('1000-a.txt')) return { Body: makeStreamBody(`${line1}\n`) }; + if (k.endsWith('2000-b.txt')) return { Body: makeStreamBody(`${line2}\n`) }; + } + return {}; }); - - const putCalls = []; - function createMockS3Client() { - return { - async send(cmd) { - if (cmd instanceof GetObjectCommand) return { Body: bodyStream }; - if (cmd instanceof PutObjectCommand) { - putCalls.push(cmd.input); - return { $metadata: { httpStatusCode: 200 } }; - } - return { $metadata: { httpStatusCode: 200 } }; - }, - }; - } - - const { writeAuditEntry, AUDIT_TIME_WINDOW_MS } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: createMockS3Client, - GetObjectCommand, - PutObjectCommand, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - const env = {}; - const ctx = { bucket: 'bkt', org: 'org1' }; - const newEntry = { - timestamp: String(1000 + AUDIT_TIME_WINDOW_MS + 1), - users: '[{"email":"a@b.com"}]', - path: 'repo/path.html', - }; - - const result = await writeAuditEntry(env, ctx, 'repo', 'file-id-1', newEntry); - - assert.strictEqual(result.status, 200); - assert.strictEqual(putCalls.length, 1); - const putBody = putCalls[0].Body; - assert.strictEqual(typeof putBody, 'string'); - const lines = putBody.split('\n').filter((l) => l.trim()); - assert.strictEqual(lines.length, 2, 'must append: existing line + new line (would fail if stream not read)'); - assert.ok(lines[0].startsWith('1000\t')); - assert.ok(lines[1].startsWith(String(1000 + AUDIT_TIME_WINDOW_MS + 1))); + const out = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); + assert.strictEqual(listCalls, 2, 'must call ListObjectsV2 twice (second uses ContinuationToken)'); + assert.strictEqual(out.length, 2); + assert.deepStrictEqual(out.map((l) => l.timestamp), [1000, 2000]); }); - - it('overwrites last line when same user and within time window', async () => { - const existingLine = '1000\t[{"email":"x@y.com"}]\trepo/f.html'; - const bodyStream = new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode(`${existingLine}\n`)); - controller.close(); - }, + it('collapses consecutive same-user edits within AUDIT_TIME_WINDOW_MS to the later entry', async () => { + const { AUDIT_TIME_WINDOW_MS } = await import(AUDIT_MODULE); + const half = Math.floor(AUDIT_TIME_WINDOW_MS / 2); + const t1 = 1000; + const t2 = t1 + half; + const line1 = `${String(t1)}\t[{"email":"u@x.com"}]\t/doc.html\t\t`; + const line2 = `${String(t2)}\t[{"email":"u@x.com"}]\t/doc.html\t\t`; + const { readAuditLines } = await mockAudit(async (cmd) => { + if (cmd instanceof ListObjectsV2Command) { + return { + Contents: [ + { Key: `o/repo/.da-versions/fid/audit/${t1}-aaaa.txt` }, + { Key: `o/repo/.da-versions/fid/audit/${t2}-bbbb.txt` }, + ], + }; + } + if (cmd instanceof GetObjectCommand) { + const k = cmd.input.Key; + if (k.endsWith('-aaaa.txt')) return { Body: makeStreamBody(`${line1}\n`) }; + return { Body: makeStreamBody(`${line2}\n`) }; + } + return {}; }); - - const putCalls = []; - function createMockS3ClientOverwrite() { - return { - async send(cmd) { - if (cmd instanceof GetObjectCommand) return { Body: bodyStream }; - if (cmd instanceof PutObjectCommand) { - putCalls.push(cmd.input); - return { $metadata: { httpStatusCode: 200 } }; - } - return { $metadata: { httpStatusCode: 200 } }; - }, - }; - } - - const { writeAuditEntry, AUDIT_TIME_WINDOW_MS } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: createMockS3ClientOverwrite, - GetObjectCommand, - PutObjectCommand, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - const newEntry = { - timestamp: String(1000 + Math.floor(AUDIT_TIME_WINDOW_MS / 2)), - users: '[{"email":"x@y.com"}]', - path: 'repo/f.html', - }; - - await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', newEntry); - - assert.strictEqual(putCalls.length, 1); - const lines = putCalls[0].Body.split('\n').filter((l) => l.trim()); - assert.strictEqual(lines.length, 1, 'must overwrite last line (same user, within window)'); + const out = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); + assert.strictEqual(out.length, 1, 'two same-user same-window edits must collapse to one'); + assert.strictEqual(out[0].timestamp, t2, 'collapse must keep the later entry'); }); - it('handles body with text() method (fetch Response-like body)', async () => { - const textBody = { - text: async () => '5000\t[{"email":"t@t.com"}]\trepo/doc.html\t\t\n', - }; - - const putCalls = []; - const { writeAuditEntry } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof GetObjectCommand) return { Body: textBody }; - if (cmd instanceof PutObjectCommand) { - putCalls.push(cmd.input); - return { $metadata: { httpStatusCode: 200 } }; - } - return { $metadata: { httpStatusCode: 200 } }; - }; - }, - GetObjectCommand, - PutObjectCommand, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - const newEntry = { - timestamp: '9999', - users: '[{"email":"t@t.com"}]', - path: 'repo/doc.html', - }; - const result = await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', newEntry); - - assert.strictEqual(result.status, 200); - assert.strictEqual(putCalls.length, 1); - // text() body was read: existing line is present (same user, within window → collapsed) - const lines = putCalls[0].Body.split('\n').filter((l) => l.trim()); - assert.ok(lines.length >= 1, 'body was read from text() stream'); + it('does NOT collapse when consecutive edits cross AUDIT_TIME_WINDOW_MS', async () => { + const { AUDIT_TIME_WINDOW_MS } = await import(AUDIT_MODULE); + const t1 = 1000; + const t2 = t1 + AUDIT_TIME_WINDOW_MS + 1; + const line1 = `${String(t1)}\t[{"email":"u@x.com"}]\t/doc.html\t\t`; + const line2 = `${String(t2)}\t[{"email":"u@x.com"}]\t/doc.html\t\t`; + const { readAuditLines } = await mockAudit(async (cmd) => { + if (cmd instanceof ListObjectsV2Command) { + return { + Contents: [ + { Key: `o/repo/.da-versions/fid/audit/${t1}-aaaa.txt` }, + { Key: `o/repo/.da-versions/fid/audit/${t2}-bbbb.txt` }, + ], + }; + } + if (cmd instanceof GetObjectCommand) { + const k = cmd.input.Key; + if (k.endsWith('-aaaa.txt')) return { Body: makeStreamBody(`${line1}\n`) }; + return { Body: makeStreamBody(`${line2}\n`) }; + } + return {}; + }); + const out = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); + assert.strictEqual(out.length, 2, 'edits outside the window must NOT collapse'); + assert.deepStrictEqual(out.map((l) => l.timestamp), [t1, t2]); }); - it('returns status 500 when GET throws a non-404 error in writeAuditEntry', async () => { - const serverError = Object.assign(new Error('server error'), { - $metadata: { httpStatusCode: 500 }, + it('does NOT collapse across different users', async () => { + const t1 = 1000; + const t2 = 2000; + const line1 = `${String(t1)}\t[{"email":"a@x.com"}]\t/doc.html\t\t`; + const line2 = `${String(t2)}\t[{"email":"b@x.com"}]\t/doc.html\t\t`; + const { readAuditLines } = await mockAudit(async (cmd) => { + if (cmd instanceof ListObjectsV2Command) { + return { + Contents: [ + { Key: `o/repo/.da-versions/fid/audit/${t1}-a.txt` }, + { Key: `o/repo/.da-versions/fid/audit/${t2}-b.txt` }, + ], + }; + } + if (cmd instanceof GetObjectCommand) { + const k = cmd.input.Key; + if (k.endsWith('-a.txt')) return { Body: makeStreamBody(`${line1}\n`) }; + return { Body: makeStreamBody(`${line2}\n`) }; + } + return {}; }); - - const { writeAuditEntry } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof GetObjectCommand) throw serverError; - return { $metadata: { httpStatusCode: 200 } }; - }; - }, - GetObjectCommand, - PutObjectCommand, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - const result = await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { - timestamp: '1000', - users: '[{"email":"x@x.com"}]', - path: 'repo/f.html', + const out = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); + assert.strictEqual(out.length, 2, 'different users must NOT collapse'); + assert.deepStrictEqual(out.map((l) => l.users[0].email), ['a@x.com', 'b@x.com']); + }); + it('version entry breaks the collapse window (edit, version, edit produces 3 entries)', async () => { + const t1 = 1000; + const t2 = 2000; + const t3 = 3000; + const edit1 = `${String(t1)}\t[{"email":"u@x.com"}]\t/doc.html\t\t`; + const ver = `${String(t2)}\t[{"email":"u@x.com"}]\t/doc.html\tRelease 1\tuuid.html`; + const edit2 = `${String(t3)}\t[{"email":"u@x.com"}]\t/doc.html\t\t`; + const { readAuditLines } = await mockAudit(async (cmd) => { + if (cmd instanceof ListObjectsV2Command) { + return { + Contents: [ + { Key: `o/repo/.da-versions/fid/audit/${t1}-a.txt` }, + { Key: `o/repo/.da-versions/fid/audit/${t2}-b.txt` }, + { Key: `o/repo/.da-versions/fid/audit/${t3}-c.txt` }, + ], + }; + } + if (cmd instanceof GetObjectCommand) { + const k = cmd.input.Key; + if (k.includes(String(t1))) return { Body: makeStreamBody(`${edit1}\n`) }; + if (k.includes(String(t2))) return { Body: makeStreamBody(`${ver}\n`) }; + return { Body: makeStreamBody(`${edit2}\n`) }; + } + return {}; }); - - assert.strictEqual(result.status, 500); - assert.strictEqual(result.error, 'server error'); + const out = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); + assert.strictEqual(out.length, 3, 'version entry must break the collapse window'); + assert.deepStrictEqual(out.map((l) => l.timestamp), [t1, t2, t3]); + assert.strictEqual(out[1].versionLabel, 'Release 1'); + assert.strictEqual(out[1].versionId, 'uuid.html'); }); - it('appends three entries when edit then version then edit (version breaks time window)', async () => { - const baseMs = 1000; - const twoMinMs = 2 * 60 * 1000; - const seventeenMinMs = 17 * 60 * 1000; - const edit1 = `${baseMs}\t[{"email":"u@x.com"}]\trepo/doc.html\t\t`; - const versionAt = `${baseMs + twoMinMs}\t[{"email":"u@x.com"}]\trepo/doc.html\tRelease 1\tuuid.html`; - const existingText = `${edit1}\n${versionAt}\n`; - const bodyStream = new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode(existingText)); - controller.close(); - }, + it('treats malformed users JSON as opaque string (collapse falls back to raw comparison)', async () => { + const t1 = 1000; + const t2 = 2000; + const line1 = `${String(t1)}\tnot-valid-json\t/doc.html\t\t`; + const line2 = `${String(t2)}\tnot-valid-json\t/doc.html\t\t`; + const { readAuditLines } = await mockAudit(async (cmd) => { + if (cmd instanceof ListObjectsV2Command) { + return { + Contents: [ + { Key: `o/repo/.da-versions/fid/audit/${t1}-a.txt` }, + { Key: `o/repo/.da-versions/fid/audit/${t2}-b.txt` }, + ], + }; + } + if (cmd instanceof GetObjectCommand) { + const k = cmd.input.Key; + if (k.endsWith('-a.txt')) return { Body: makeStreamBody(`${line1}\n`) }; + return { Body: makeStreamBody(`${line2}\n`) }; + } + return {}; }); - - const putCalls = []; - const mockSend = (cmd) => { - if (cmd instanceof GetObjectCommand) return { Body: bodyStream }; + const out = await readAuditLines({}, { bucket: 'b', org: 'o' }, 'repo', 'fid'); + assert.strictEqual(out.length, 1, 'identical malformed-JSON users normalize to the same raw string and collapse'); + assert.strictEqual(out[0].timestamp, t2); + }); + }); + describe('writeAuditEntry (append-only ledger)', () => { + it('writes a single unconditional PUT to a fresh per-entry key (no GET, no If-Match, no retry)', async () => { + const calls = []; + const { writeAuditEntry } = await mockAudit(async (cmd) => { + if (cmd instanceof GetObjectCommand) { + throw new Error('writeAuditEntry must not GET on the append-only path'); + } if (cmd instanceof PutObjectCommand) { - putCalls.push(cmd.input); + calls.push(cmd.input); return { $metadata: { httpStatusCode: 200 } }; } return { $metadata: { httpStatusCode: 200 } }; - }; - - const { writeAuditEntry } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { this.send = mockSend; }, - GetObjectCommand, - PutObjectCommand, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - const edit2At = baseMs + seventeenMinMs; - await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { - timestamp: String(edit2At), - users: '[{"email":"u@x.com"}]', - path: 'repo/doc.html', - }); - - assert.strictEqual(putCalls.length, 1); - const lines = putCalls[0].Body.split('\n').filter((l) => l.trim()); - assert.strictEqual(lines.length, 3, 'edit at 12:23, version at 12:25, edit at 12:40 => 3 entries'); - assert.ok(lines[0].endsWith('\t\t'), 'first line is edit (no version)'); - assert.ok(lines[1].includes('Release 1') && lines[1].includes('uuid.html'), 'second line is version'); - assert.ok(lines[2].startsWith(String(edit2At)) && lines[2].endsWith('\t\t'), 'third line is edit'); - }); - - it('sends If-Match header on PUT using ETag from GET', async () => { - const bodyStream = new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode('')); - controller.close(); - }, }); - - const putCalls = []; - const { writeAuditEntry } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof GetObjectCommand) { - return { Body: bodyStream, ETag: '"etag-abc"' }; - } - if (cmd instanceof PutObjectCommand) { - putCalls.push(cmd.input); - return { $metadata: { httpStatusCode: 200 } }; - } - return { $metadata: { httpStatusCode: 200 } }; - }; - }, - GetObjectCommand, - PutObjectCommand, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { - timestamp: '1000', + const result = await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { + timestamp: '5000', users: '[{"email":"u@x.com"}]', path: 'repo/doc.html', }); - - assert.strictEqual(putCalls.length, 1); - assert.strictEqual(putCalls[0].IfMatch, '"etag-abc"', 'If-Match must equal ETag from GET'); + assert.strictEqual(result.status, 200); + assert.strictEqual(calls.length, 1, 'exactly one PUT'); + assert.strictEqual(calls[0].IfMatch, undefined, 'append-only PUT must not send If-Match'); + assert.match(calls[0].Key, /^o\/repo\/\.da-versions\/fid\/audit\/5000-[a-f0-9]{16}\.txt$/); + const lns = calls[0].Body.split('\n').filter((l) => l.trim()); + assert.strictEqual(lns.length, 1, 'body is exactly one formatted entry line'); + assert.ok(lns[0].startsWith('5000\t')); + assert.ok(lns[0].includes('repo/doc.html')); }); - it('omits If-Match when file does not yet exist (first write)', async () => { - const putCalls = []; - const { writeAuditEntry } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof GetObjectCommand) { - const err = new Error('not found'); - err.name = 'NoSuchKey'; - throw err; - } - if (cmd instanceof PutObjectCommand) { - putCalls.push(cmd.input); - return { $metadata: { httpStatusCode: 200 } }; - } - return { $metadata: { httpStatusCode: 200 } }; - }; - }, - GetObjectCommand, - PutObjectCommand, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { - timestamp: '1000', + it('does NOT retry on 412 from PUT (append-only: no etag, no contention)', async () => { + let putCalls = 0; + const { writeAuditEntry } = await mockAudit(async (cmd) => { + if (cmd instanceof PutObjectCommand) { + putCalls += 1; + const err = new Error('precondition failed'); + err.$metadata = { httpStatusCode: 412 }; + throw err; + } + return { $metadata: { httpStatusCode: 200 } }; + }); + const result = await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { + timestamp: '5000', users: '[{"email":"u@x.com"}]', path: 'repo/doc.html', }); - - assert.strictEqual(putCalls.length, 1); - assert.strictEqual(putCalls[0].IfMatch, undefined, 'If-Match must be absent for first write'); + assert.strictEqual(result.status, 500, '412 must surface as 500 immediately with no retry'); + assert.strictEqual(putCalls, 1, 'append-only path attempts PUT exactly once'); }); - it('retries on 412 from PUT and succeeds on a later attempt', async () => { - let getCallCount = 0; - const putCalls = []; - - const makeBody = () => new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode('')); - controller.close(); - }, + it('returns 500 with error message when PUT throws', async () => { + const { writeAuditEntry } = await mockAudit(async (cmd) => { + if (cmd instanceof PutObjectCommand) { + throw new Error('network down'); + } + return { $metadata: { httpStatusCode: 200 } }; }); - - const { writeAuditEntry } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof GetObjectCommand) { - getCallCount += 1; - return { Body: makeBody(), ETag: `"etag-${getCallCount}"` }; - } - if (cmd instanceof PutObjectCommand) { - putCalls.push(cmd.input); - if (putCalls.length < 3) { - // First two PUTs: simulate concurrent write → 412 - const err = new Error('precondition failed'); - err.$metadata = { httpStatusCode: 412 }; - throw err; - } - return { $metadata: { httpStatusCode: 200 } }; - } - return { $metadata: { httpStatusCode: 200 } }; - }; - }, - GetObjectCommand, - PutObjectCommand, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - const result = await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { timestamp: '5000', users: '[{"email":"u@x.com"}]', path: 'repo/doc.html', }); - - assert.strictEqual(result.status, 200); - assert.strictEqual(getCallCount, 3, 'must re-read on each retry'); - assert.strictEqual(putCalls.length, 3, 'must retry the PUT until success'); - assert.strictEqual(putCalls[0].IfMatch, '"etag-1"'); - assert.strictEqual(putCalls[1].IfMatch, '"etag-2"', 'first retry uses fresh ETag'); - assert.strictEqual(putCalls[2].IfMatch, '"etag-3"', 'second retry uses fresh ETag'); + assert.strictEqual(result.status, 500); + assert.strictEqual(result.error, 'network down'); }); - - it('archives existing content and starts fresh when entry count reaches AUDIT_MAX_ENTRIES', async () => { - const { AUDIT_MAX_ENTRIES } = await import('../../../src/storage/version/audit.js'); - - const lastTs = 5000; - const existingLines = Array.from({ length: AUDIT_MAX_ENTRIES }, (_, i) => ( - `${1000 + i}\t[{"email":"u@x.com"}]\t/doc.html\t\t` - )); - existingLines[existingLines.length - 1] = `${lastTs}\t[{"email":"u@x.com"}]\t/doc.html\t\t`; - const existingText = `${existingLines.join('\n')}\n`; - - const putCalls = []; - const mockSend = async (cmd) => { - if (cmd instanceof GetObjectCommand) { - return { - Body: new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode(existingText)); - controller.close(); - }, - }), - ETag: '"etag-1"', - }; - } + it('uses Date.now() as fallback when entry.timestamp is not numeric', async () => { + const calls = []; + const { writeAuditEntry } = await mockAudit(async (cmd) => { if (cmd instanceof PutObjectCommand) { - putCalls.push(cmd.input); + calls.push(cmd.input); return { $metadata: { httpStatusCode: 200 } }; } - return {}; - }; - - const { writeAuditEntry } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { this.send = mockSend; }, - GetObjectCommand, - PutObjectCommand, - ListObjectsV2Command, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - const newEntry = { - timestamp: String(lastTs + 10000000), - users: '[{"email":"other@x.com"}]', - path: '/doc.html', - }; - - const result = await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', newEntry); - - assert.strictEqual(result.status, 200); - assert.strictEqual(putCalls.length, 2, 'must PUT archive + new audit.txt'); - - const archivePut = putCalls.find((p) => p.Key.includes('audit-')); - const auditPut = putCalls.find((p) => p.Key.endsWith('audit.txt')); - assert.ok(archivePut, 'archive PUT must exist'); - assert.ok(auditPut, 'audit.txt PUT must exist'); - assert.strictEqual(archivePut.Body, existingText, 'archive must contain the old content'); - assert.ok(archivePut.Key.includes(`audit-${lastTs}`), 'archive key must use last entry timestamp'); - const newAuditLines = auditPut.Body.split('\n').filter((l) => l.trim()); - assert.strictEqual(newAuditLines.length, 1, 'new audit.txt must contain only the new entry'); - assert.ok(newAuditLines[0].includes(newEntry.timestamp), 'new entry must be present in fresh audit.txt'); - }); - - it('treats malformed users JSON in existing entry as opaque string (no crash)', async () => { - // Covers usersNormalized catch branch: invalid JSON falls back to raw string comparison. - const existingLine = '1000\tnot-valid-json\t/doc.html\t\t'; - const putCalls = []; - - const { writeAuditEntry } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof GetObjectCommand) { - return { - Body: new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode(`${existingLine}\n`)); - controller.close(); - }, - }), - ETag: '"etag-bad"', - }; - } - if (cmd instanceof PutObjectCommand) { - putCalls.push(cmd.input); - return { $metadata: { httpStatusCode: 200 } }; - } - return {}; - }; - }, - GetObjectCommand, - PutObjectCommand, - ListObjectsV2Command, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - // Different user — must not collapse with the malformed-JSON entry - const result = await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { - timestamp: '9000', + return { $metadata: { httpStatusCode: 200 } }; + }); + const before = Date.now(); + await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { + timestamp: 'not-a-number', users: '[{"email":"u@x.com"}]', - path: '/doc.html', + path: 'repo/doc.html', }); - - assert.strictEqual(result.status, 200); - assert.strictEqual(putCalls.length, 1); - const lines = putCalls[0].Body.split('\n').filter((l) => l.trim()); - assert.strictEqual(lines.length, 2, 'both entries must be present (no collapse across mismatched users)'); + const after = Date.now(); + const m = calls[0].Key.match(/audit\/(\d+)-[a-f0-9]{16}\.txt$/); + assert.ok(m, 'key embeds numeric timestamp'); + const ts = parseInt(m[1], 10); + assert.ok(ts >= before && ts <= after, 'timestamp falls back to Date.now() when entry value is non-numeric'); + const lineTs = calls[0].Body.split('\t')[0]; + assert.strictEqual(lineTs, String(ts), 'serialized entry timestamp matches key timestamp'); }); - it('returns status 500 when PUT 412 persists across all 7 attempts', async () => { - let getCallCount = 0; - let putCallCount = 0; - - const makeBody = () => new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode('')); - controller.close(); - }, + it('generates a fresh random suffix per call (two concurrent writes do not collide on key)', async () => { + const calls = []; + const { writeAuditEntry } = await mockAudit(async (cmd) => { + if (cmd instanceof PutObjectCommand) { + calls.push(cmd.input); + return { $metadata: { httpStatusCode: 200 } }; + } + return { $metadata: { httpStatusCode: 200 } }; }); - - const { writeAuditEntry } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof GetObjectCommand) { - getCallCount += 1; - return { Body: makeBody(), ETag: '"etag-x"' }; - } - if (cmd instanceof PutObjectCommand) { - putCallCount += 1; - const err = new Error('precondition failed'); - err.$metadata = { httpStatusCode: 412 }; - throw err; - } - return { $metadata: { httpStatusCode: 200 } }; - }; - }, - GetObjectCommand, - PutObjectCommand, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - const result = await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { + const entry = { timestamp: '5000', users: '[{"email":"u@x.com"}]', path: 'repo/doc.html', - }); - - assert.strictEqual(result.status, 500, 'persistent 412 must surface as 500 after 7 total attempts'); - assert.strictEqual(result.error, 'precondition failed'); - assert.strictEqual(putCallCount, 7, 'must attempt PUT 7 times total (1 initial + 6 retries)'); - assert.strictEqual(getCallCount, 7, 'must re-read on each attempt'); - }); - - it('uses exponential jitter backoff (0-50, 0-100, 0-200, 0-400, 0-800, 0-1600 ms) across six 412 retries', async () => { - // Stubs setTimeout and Math.random to capture per-retry delays. - // Asserts per-attempt exponential upper bounds for the 6-retry ladder - // (50, 100, 200, 400, 800, 1600 ms) and total elapsed sleep ~3050 ms. - const originalSetTimeout = globalThis.setTimeout; - const originalRandom = Math.random; - const delays = []; - Math.random = () => 0.999999; - globalThis.setTimeout = (fn, ms) => { - delays.push(ms); - fn(); - return 0; }; - - try { - const makeBody = () => new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode('')); - controller.close(); - }, - }); - - const { writeAuditEntry } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof GetObjectCommand) { - return { Body: makeBody(), ETag: '"etag-x"' }; - } - if (cmd instanceof PutObjectCommand) { - const err = new Error('precondition failed'); - err.$metadata = { httpStatusCode: 412 }; - throw err; - } - return { $metadata: { httpStatusCode: 200 } }; - }; - }, - GetObjectCommand, - PutObjectCommand, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { - timestamp: '5000', - users: '[{"email":"u@x.com"}]', - path: 'repo/doc.html', - }); - } finally { - globalThis.setTimeout = originalSetTimeout; - Math.random = originalRandom; - } - - assert.strictEqual(delays.length, 6, 'must sleep between each of the 6 retries'); - const upperBounds = [50, 100, 200, 400, 800, 1600]; - delays.forEach((ms, i) => { - assert.ok( - ms >= 0 && ms < upperBounds[i], - `delay[${i}]=${ms} must be within [0, ${upperBounds[i]})`, - ); - }); - assert.ok( - delays[2] >= 150, - `delay[2]=${delays[2]} must exceed the prior linear cap of 150 ms`, - ); - assert.ok( - delays[3] >= 200, - `delay[3]=${delays[3]} must exceed the prior linear cap of 200 ms`, - ); - assert.ok( - delays[5] >= 800, - `delay[5]=${delays[5]} must reach the new 6-retry exponential ceiling (>=800 ms)`, - ); - const total = delays.reduce((a, b) => a + b, 0); - assert.ok( - total > 1500, - `total elapsed sleep ${total} must exceed prior 4-retry worst-case (~750 ms)`, - ); - assert.ok( - total < 3200, - `total elapsed sleep ${total} must stay within the new 6-retry worst-case (~3050 ms)`, - ); + await Promise.all([ + writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', entry), + writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', entry), + ]); + assert.strictEqual(calls.length, 2); + assert.notStrictEqual(calls[0].Key, calls[1].Key, 'concurrent writes must produce distinct keys'); }); - it('succeeds on the 5th attempt after four 412s', async () => { - let getCallCount = 0; - let putCallCount = 0; - - const makeBody = () => new ReadableStream({ - start(controller) { - controller.enqueue(new TextEncoder().encode('')); - controller.close(); - }, + it('writes version entry (versionLabel + versionId) to its own per-entry object', async () => { + const calls = []; + const { writeAuditEntry } = await mockAudit(async (cmd) => { + if (cmd instanceof PutObjectCommand) { + calls.push(cmd.input); + return { $metadata: { httpStatusCode: 200 } }; + } + return { $metadata: { httpStatusCode: 200 } }; }); - - const { writeAuditEntry } = await esmock( - '../../../src/storage/version/audit.js', - { - '@aws-sdk/client-s3': { - S3Client: function S3Client() { - this.send = async (cmd) => { - if (cmd instanceof GetObjectCommand) { - getCallCount += 1; - return { Body: makeBody(), ETag: `"etag-${getCallCount}"` }; - } - if (cmd instanceof PutObjectCommand) { - putCallCount += 1; - if (putCallCount < 5) { - const err = new Error('precondition failed'); - err.$metadata = { httpStatusCode: 412 }; - throw err; - } - return { $metadata: { httpStatusCode: 200 } }; - } - return { $metadata: { httpStatusCode: 200 } }; - }; - }, - GetObjectCommand, - PutObjectCommand, - }, - '../../../src/storage/utils/config.js': { default: () => ({}) }, - }, - ); - - const result = await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { + await writeAuditEntry({}, { bucket: 'b', org: 'o' }, 'repo', 'fid', { timestamp: '5000', users: '[{"email":"u@x.com"}]', path: 'repo/doc.html', + versionLabel: 'Release 1', + versionId: 'uuid', }); - - assert.strictEqual(result.status, 200, 'must succeed on the 5th attempt'); - assert.strictEqual(putCallCount, 5, 'must have attempted PUT 5 times'); - assert.strictEqual(getCallCount, 5, 'must re-read on each attempt'); + assert.strictEqual(calls.length, 1); + assert.ok(calls[0].Body.includes('Release 1')); + assert.ok(calls[0].Body.includes('uuid')); }); }); }); diff --git a/test/storage/version/paths.test.js b/test/storage/version/paths.test.js index 0e03f539..a6ae3cdd 100644 --- a/test/storage/version/paths.test.js +++ b/test/storage/version/paths.test.js @@ -15,6 +15,7 @@ import { auditKey, auditArchiveKey, auditDirPrefix, + auditEntryKey, } from '../../../src/storage/version/paths.js'; describe('Version Paths', () => { @@ -57,9 +58,30 @@ describe('Version Paths', () => { }); describe('auditDirPrefix', () => { - it('returns prefix that matches audit.txt and audit-*.txt', () => { + it('returns prefix that matches audit.txt, audit-*.txt, and per-entry objects', () => { const prefix = auditDirPrefix('myrepo', 'file-id-xyz'); assert.strictEqual(prefix, 'myrepo/.da-versions/file-id-xyz/audit'); }); }); + + describe('auditEntryKey', () => { + it('returns per-entry object key with ts-rand suffix under the audit/ prefix', () => { + const key = auditEntryKey('myrepo', 'file-id-xyz', 1234567890, 'deadbeefcafef00d'); + assert.strictEqual(key, 'myrepo/.da-versions/file-id-xyz/audit/1234567890-deadbeefcafef00d.txt'); + }); + + it('accepts string timestamp as-is', () => { + assert.strictEqual( + auditEntryKey('r', 'fid', '9999', 'abcd1234'), + 'r/.da-versions/fid/audit/9999-abcd1234.txt', + ); + }); + + it('lives under the same prefix as auditDirPrefix (so ListObjectsV2 picks it up alongside legacy files)', () => { + const repo = 'myrepo'; + const fileId = 'file-1'; + const prefix = auditDirPrefix(repo, fileId); + assert.ok(auditEntryKey(repo, fileId, 1000, 'r1').startsWith(prefix)); + }); + }); }); diff --git a/test/storage/version/put.test.js b/test/storage/version/put.test.js index df42459b..13acd64b 100644 --- a/test/storage/version/put.test.js +++ b/test/storage/version/put.test.js @@ -2634,7 +2634,7 @@ describe('Version Put', () => { ); }); - it('calls writeAuditEntry once and succeeds (retries are handled inside writeAuditEntry)', async () => { + it('calls writeAuditEntry exactly once per versionable PUT (append-only ledger: no retries)', async () => { let callCount = 0; const mockWriteAuditEntry = async () => { callCount += 1; @@ -2670,7 +2670,7 @@ describe('Version Put', () => { ); assert.strictEqual(resp.status, 200, 'document write must succeed'); - assert.strictEqual(callCount, 1, 'put.js must call writeAuditEntry exactly once (retries are handled inside writeAuditEntry)'); + assert.strictEqual(callCount, 1, 'put.js must call writeAuditEntry exactly once (append-only ledger: a single unconditional PUT inside, no retries)'); }); it('writeAuditEntry returning status 500 does not affect main put result', async () => {