Skip to content

fix(audit): append-only ledger eliminates writeAuditEntry contention #285

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
kptdobe wants to merge 1 commit into from
+487 −969
Closed

fix(audit): append-only ledger eliminates writeAuditEntry contention #285

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
196 changes: 81 additions & 115 deletions src/storage/version/audit.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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';

/**
Expand Down Expand Up @@ -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<string>}
*/
async function streamToString(body) {
Expand Down Expand Up @@ -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<object[]>} 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 [];
Expand All @@ -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);
Expand All @@ -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);
Expand Down
16 changes: 15 additions & 1 deletion src/storage/version/paths.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,11 +44,25 @@ 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)
*/
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`;
}
9 changes: 3 additions & 6 deletions test/storage/object/conditionals.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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);
});
});

Expand Down
Loading
Loading