From 6413db9ae3d09834036ff0d2198b664bafdee692 Mon Sep 17 00:00:00 2001 From: Chidubemkingsley Date: Sun, 27 Sep 2026 22:28:48 +0100 Subject: [PATCH] fix(signals): stop losing writes on concurrent signal mutations The signal store did readFile -> JSON.parse -> mutate -> writeFile on a shared JSON sidecar with no locking and no version. Under concurrency every writer read the same snapshot and the last writeFile won, so writers silently erased each other. Measured with 50 concurrent writers: 49 of 50 signals, upvotes, and replies were lost. Three layers, all in lib/signal-store.ts: - Serialized read-modify-write via a per-file promise queue, so two mutations cannot both read the same snapshot. - Atomic replacement: write to a sibling .tmp file, then rename(2), so readers never observe a half-flushed file and a crash mid-write cannot corrupt the live store. - A JSON parse failure now throws instead of degrading to {}. Returning {} on a parse error meant the next write overwrote every record. Each Signal gains an integer version, bumped on every mutation: - GET /api/signals/[id] publishes it as an ETag. - POST /api/signals/[id] accepts an optional If-Match and returns 409 with current_version plus the fresh ETag on a stale value. - POST /api/signals/[id]/replies accepts an optional parent_version and returns 409 on a stale value. - Conflicts log under the signals.version_conflict event. Both version inputs are optional, so clients that omit them keep the previous last-writer-wins behaviour. Records written before this change have no version and are normalized to 1 on read, so there is no migration step. Adds the project's first test suite (vitest): 13 tests covering 50-concurrent-writer create/upvote/reply, CAS races, corrupt-store refusal, legacy version backfill, and version headers. Verified the concurrency tests fail against the old store before passing against the new one. Residual limitation, documented in docs/TECHNICAL.md 5.3: the queue is per-process, so two Vercel instances can still interleave. The other six stores (follow, profile, market, notification, achievement, narrative-world) share the same unsafe pattern and are untouched here. --- PROJECT_ARCHITECTURE.md | 7 + app/api/signals/[id]/replies/route.ts | 49 +- app/api/signals/[id]/route.ts | 92 +- app/signals/[id]/page.tsx | 6 +- app/signals/[id]/signal-detail-client.tsx | 31 +- docs/TECHNICAL.md | 51 +- .../signal-store-concurrency.test.ts | 242 +++ lib/signal-store.ts | 190 +- package-lock.json | 1844 ++++++++++++++++- package.json | 6 +- vitest.config.ts | 14 + 11 files changed, 2400 insertions(+), 132 deletions(-) create mode 100644 lib/__tests__/signal-store-concurrency.test.ts create mode 100644 vitest.config.ts diff --git a/PROJECT_ARCHITECTURE.md b/PROJECT_ARCHITECTURE.md index 9bc35567..fbe44e43 100644 --- a/PROJECT_ARCHITECTURE.md +++ b/PROJECT_ARCHITECTURE.md @@ -52,6 +52,7 @@ Owns: - Trustline submit relay endpoint (for signed XDR). - Persistent JSON data (`faucet claims`, `artist profile`, `listings`) through `lib/server-data-paths.ts`. - x402 endpoints and payment verification support. +- Concurrency safety for JSON-backed stores: serialized read-modify-write plus atomic file replacement (see `lib/signal-store.ts`). Must not own: @@ -95,6 +96,12 @@ Owns: - **Deterministic status codes** (validation, cooldown, authorization, pending). - **No implicit success**: all state transitions explicit and auditable. - **Compatibility routes** allowed if typed and documented. +- **Optimistic concurrency on shared mutable records**: a record that more than one + client can edit carries an integer `version`. Reads publish it as an `ETag`; + mutations may echo it back as `If-Match` (or `parent_version` for appends) and + the server rejects a stale value with `409` rather than silently overwriting a + concurrent writer. `409` is logged under the `signals.version_conflict` event + so conflict rates are observable. ## 6) Internationalization architecture diff --git a/app/api/signals/[id]/replies/route.ts b/app/api/signals/[id]/replies/route.ts index 89ddbc70..7db82a6f 100644 --- a/app/api/signals/[id]/replies/route.ts +++ b/app/api/signals/[id]/replies/route.ts @@ -1,6 +1,6 @@ import { NextRequest } from "next/server" import { StrKey } from "@stellar/stellar-sdk" -import { getSignal, createReply } from "@/lib/signal-store" +import { getSignal, createReply, VersionConflictError } from "@/lib/signal-store" import { createNotification } from "@/lib/notification-store" import { createApiRequestContext } from "@/lib/api-observability" @@ -11,6 +11,7 @@ type ReplyBody = { body?: unknown wallet?: unknown signature?: unknown + parent_version?: unknown } export async function POST( @@ -51,6 +52,21 @@ export async function POST( ) } + // Optional: when present, the reply must be composed against this version of + // the parent signal. A stale parent_version is rejected rather than silently + // accepted against a view the author never saw. + let parentVersion: number | undefined + if (body.parent_version !== undefined) { + const parsed = body.parent_version + if (typeof parsed !== "number" || !Number.isInteger(parsed) || parsed < 1) { + return api.json( + { error: "Invalid parent_version" }, + { status: 400, event: "signals.reply.validation_failed", metadata: { reason: "parent_version" } }, + ) + } + parentVersion = parsed + } + try { const signal = await getSignal(id) if (!signal) { @@ -73,14 +89,17 @@ export async function POST( } } - const reply = await createReply({ - signal_id: id, - author_wallet: walletStr, - author_display, - body: (body.body as string).trim(), - upvotes: [], - signature: body.signature as string, - }) + const reply = await createReply( + { + signal_id: id, + author_wallet: walletStr, + author_display, + body: (body.body as string).trim(), + upvotes: [], + signature: body.signature as string, + }, + parentVersion, + ) if (signal.author_wallet !== walletStr) { void createNotification(signal.author_wallet, "signal_reply", { @@ -96,6 +115,16 @@ export async function POST( { status: 201, event: "signals.reply.created", metadata: { signal_id: id, reply_id: reply.id } }, ) } catch (error) { + if (error instanceof VersionConflictError) { + return api.json( + { error: "Version conflict", current_version: error.currentVersion }, + { + status: 409, + event: "signals.version_conflict", + metadata: { signal_id: id, route: "replies", current_version: error.currentVersion }, + }, + ) + } return api.errorJson(error, 500, "signals.reply.create_failed") } -} \ No newline at end of file +} diff --git a/app/api/signals/[id]/route.ts b/app/api/signals/[id]/route.ts index 5f8259d1..a36a5b4e 100644 --- a/app/api/signals/[id]/route.ts +++ b/app/api/signals/[id]/route.ts @@ -1,23 +1,38 @@ -import { NextRequest, NextResponse } from "next/server" +import { NextRequest } from "next/server" import { StrKey } from "@stellar/stellar-sdk" -import { getSignal, upvoteSignal, getReplies } from "@/lib/signal-store" +import { + getSignal, + upvoteSignal, + getReplies, + signalETag, + parseVersionHeader, + VersionConflictError, +} from "@/lib/signal-store" import { createNotification } from "@/lib/notification-store" import { checkAndUnlock } from "@/lib/achievement-store" +import { createApiRequestContext } from "@/lib/api-observability" export const runtime = "nodejs" export const dynamic = "force-dynamic" export async function GET( - _request: NextRequest, + request: NextRequest, { params }: { params: Promise<{ id: string }> }, ) { + const api = createApiRequestContext(request, "/api/signals/[id]") const { id } = await params const signal = await getSignal(id) if (!signal) { - return NextResponse.json({ error: "Signal not found" }, { status: 404 }) + return api.json( + { error: "Signal not found" }, + { status: 404, event: "signals.get.not_found", metadata: { signal_id: id } }, + ) } const replies = await getReplies(id) - return NextResponse.json({ signal, replies }) + return api.json( + { signal, replies }, + { status: 200, event: "signals.get.ok", headers: { ETag: signalETag(signal.version) } }, + ) } type UpvoteBody = { @@ -29,23 +44,48 @@ export async function POST( request: NextRequest, { params }: { params: Promise<{ id: string }> }, ) { + const api = createApiRequestContext(request, "/api/signals/[id]") const { id } = await params let body: UpvoteBody try { body = (await request.json()) as UpvoteBody } catch { - return NextResponse.json({ error: "Invalid JSON" }, { status: 400 }) + return api.json( + { error: "Invalid JSON" }, + { status: 400, event: "signals.upvote.invalid_json" }, + ) } if (typeof body.wallet !== "string" || !StrKey.isValidEd25519PublicKey(body.wallet)) { - return NextResponse.json({ error: "Invalid wallet address" }, { status: 400 }) + return api.json( + { error: "Invalid wallet address" }, + { status: 400, event: "signals.upvote.validation_failed", metadata: { reason: "wallet" } }, + ) } if (typeof body.signature !== "string" || body.signature.length === 0) { - return NextResponse.json({ error: "Signature required" }, { status: 400 }) + return api.json( + { error: "Signature required" }, + { status: 400, event: "signals.upvote.validation_failed", metadata: { reason: "signature" } }, + ) + } + + // If-Match is optional so existing clients keep working; when supplied the + // upvote becomes a compare-and-swap against the version the client rendered. + const ifMatch = request.headers.get("if-match") + let expectedVersion: number | undefined + if (ifMatch !== null && ifMatch.trim() !== "*") { + const parsed = parseVersionHeader(ifMatch) + if (parsed === null) { + return api.json( + { error: "Invalid If-Match" }, + { status: 400, event: "signals.upvote.validation_failed", metadata: { reason: "if_match" } }, + ) + } + expectedVersion = parsed } try { - const signal = await upvoteSignal(id, body.wallet) + const signal = await upvoteSignal(id, body.wallet, expectedVersion) // Notify at milestones: 5, 10, 25 upvotes (fire-and-forget) const count = signal.upvotes.length if ((count === 5 || count === 10 || count === 25) && signal.author_wallet !== body.wallet) { @@ -53,14 +93,38 @@ export async function POST( signal_id: id, signal_title: signal.title, upvote_count: count, - }).catch(() => { /* silent */ }) + }).catch((error) => api.log("warn", "signals.upvote.notification_failed", { error })) } // Achievement: track upvotes for the author (fire-and-forget) if (signal.author_wallet !== body.wallet) { - void checkAndUnlock(signal.author_wallet, { upvote_delta: 1 }).catch(() => { /* silent */ }) + void checkAndUnlock(signal.author_wallet, { upvote_delta: 1 }).catch((error) => + api.log("warn", "signals.upvote.achievement_failed", { error }), + ) } - return NextResponse.json({ signal }) - } catch { - return NextResponse.json({ error: "Signal not found" }, { status: 404 }) + return api.json( + { signal }, + { + status: 200, + event: "signals.upvote.ok", + metadata: { signal_id: id, version: signal.version }, + headers: { ETag: signalETag(signal.version) }, + }, + ) + } catch (error) { + if (error instanceof VersionConflictError) { + return api.json( + { error: "Version conflict", current_version: error.currentVersion }, + { + status: 409, + event: "signals.version_conflict", + metadata: { signal_id: id, current_version: error.currentVersion }, + headers: { ETag: signalETag(error.currentVersion) }, + }, + ) + } + return api.json( + { error: "Signal not found" }, + { status: 404, event: "signals.upvote.not_found", metadata: { signal_id: id } }, + ) } } diff --git a/app/signals/[id]/page.tsx b/app/signals/[id]/page.tsx index 521947f5..51d9d4ee 100644 --- a/app/signals/[id]/page.tsx +++ b/app/signals/[id]/page.tsx @@ -127,7 +127,11 @@ export default async function SignalDetailPage({ params }: Props) { {/* Replies + compose (client island) */} - + ) diff --git a/app/signals/[id]/signal-detail-client.tsx b/app/signals/[id]/signal-detail-client.tsx index 631d97e3..3b915a79 100644 --- a/app/signals/[id]/signal-detail-client.tsx +++ b/app/signals/[id]/signal-detail-client.tsx @@ -15,6 +15,7 @@ const copy = { ctaBusy: "[ SENDING… ]", noWallet: "[ CONNECT_WALLET_TO_REPLY ]", walletBadge: "✓ WALLET", + conflict: "[ SIGNAL_CHANGED_REFRESH_AND_RETRY ]", }, es: { replies: "RESPUESTAS", @@ -24,6 +25,7 @@ const copy = { ctaBusy: "[ ENVIANDO… ]", noWallet: "[ CONECTAR_WALLET_PARA_RESPONDER ]", walletBadge: "✓ WALLET", + conflict: "[ SEÑAL_ACTUALIZADA_REFRESCA_Y_REINTENTA ]", }, } @@ -39,18 +41,31 @@ function timeAgo(ts: number): string { type Props = { signalId: string + initialSignalVersion: number initialReplies: SignalReply[] } -export function SignalDetailClient({ signalId, initialReplies }: Props) { +type PostReplyResponse = { + reply?: SignalReply + error?: string + current_version?: number +} + +export function SignalDetailClient({ + signalId, + initialSignalVersion, + initialReplies, +}: Props) { const { address } = useWallet() const { lang } = useLang() const t = copy[lang] ?? copy.en const [replies, setReplies] = useState(initialReplies) + const [signalVersion, setSignalVersion] = useState(initialSignalVersion) const [replyBody, setReplyBody] = useState("") const [busy, setBusy] = useState(false) const [error, setError] = useState(null) + const [conflict, setConflict] = useState(false) const baseInput = "w-full bg-transparent border border-[var(--color-border-tertiary)] font-mono text-[12px] text-foreground px-3 py-2 focus:outline-none focus:border-[#7F77DD] transition-colors placeholder:text-muted-foreground/40 resize-none" @@ -58,6 +73,7 @@ export function SignalDetailClient({ signalId, initialReplies }: Props) { async function handleReply() { if (!address || !replyBody.trim()) return setError(null) + setConflict(false) setBusy(true) try { const res = await fetch(`/api/signals/${signalId}/replies`, { @@ -68,9 +84,17 @@ export function SignalDetailClient({ signalId, initialReplies }: Props) { wallet: address, // TODO: replace provisional signature with Freighter signMessage when available signature: address, + parent_version: signalVersion, }), }) - const data = (await res.json().catch(() => ({}))) as { reply?: SignalReply; error?: string } + const data = (await res.json().catch(() => ({}))) as PostReplyResponse + if (res.status === 409) { + if (typeof data.current_version === "number") { + setSignalVersion(data.current_version) + } + setConflict(true) + return + } if (!res.ok || !data.reply) { setError(data.error ?? "Reply failed") return @@ -148,6 +172,9 @@ export function SignalDetailClient({ signalId, initialReplies }: Props) { {error && (

{error}

)} + {conflict && ( +

{t.conflict}

+ )}