mirror of
https://github.com/tiennm99/rplace.git
synced 2026-10-16 00:15:17 +00:00
fix(canvas): make multi-chunk pixel writes atomic and refund cooldown on failure
- wrap writePixels in state.storage.transactionSync so a partial multi-chunk failure doesn't leave the canvas half-written or out of sync with the WS broadcast - size new chunk buffer against chunkSize(chunkId) instead of the persisted blob's length so writes after a canvas-grow no longer silently drop OOB bytes in the formerly-last short chunk - refund the cooldown row when writePixels throws so transient storage errors stop soft-DOSing the user (and halving image-uploader throughput) - bound readAllChunks by chunk_id < CHUNK_COUNT and trim oversized blobs so orphan rows from a future shrink no longer crash GET /api/canvas - require a positive Content-Length on /api/place (411) and reject above the pre-parse cap (413); previously a missing or zero header bypassed the cap - drop String(err) from the 500 response body - drain the INSERT cursor symmetrically with the UPDATE branch in tryAcquire - assert CHUNK_BYTES <= 2 MB at module load (DO SQLite per-cell BLOB cap) - correct the inverted webSocketClose comment and guard the re-close call - add tests for missing / zero / oversized Content-Length Plan: plans/260510-0232-fix-do-migration-followups/phase-01-do-storage-atomicity.md
This commit is contained in:
1 parent
a977adc62d
commit
3c1263add6
15 files changed
+1686
-24
No files matched your search
@@ -1,7 +1,7 @@
|
||||
import { CANVAS_WIDTH, CANVAS_HEIGHT, MAX_COLORS, MAX_BATCH_SIZE } from '../lib/constants.js';
|
||||
import { init as initSchema } from './lib/schema.js';
|
||||
import { readAllChunks, writePixels } from './lib/chunk-storage.js';
|
||||
import { tryAcquire } from './lib/cooldown-store.js';
|
||||
import { tryAcquire, release } from './lib/cooldown-store.js';
|
||||
|
||||
/**
|
||||
* CanvasRoom: single Durable Object that owns
|
||||
@@ -76,10 +76,20 @@ export class CanvasRoom {
|
||||
}
|
||||
|
||||
try {
|
||||
writePixels(this.sql, pixels);
|
||||
// transactionSync makes the multi-chunk batch all-or-nothing. Without
|
||||
// it, each sql.exec auto-commits and a partial failure leaves the
|
||||
// canvas in a half-written state.
|
||||
this.state.storage.transactionSync(() => {
|
||||
writePixels(this.sql, pixels);
|
||||
});
|
||||
} catch (err) {
|
||||
console.error('writePixels failed:', err);
|
||||
return Response.json({ error: 'storage_failed', message: String(err) }, { status: 500 });
|
||||
// Refund the cooldown so a transient storage error doesn't lock the
|
||||
// user out for 1s (and the image-uploader doesn't lose throughput).
|
||||
try { release(this.sql, userId); } catch (releaseErr) {
|
||||
console.warn('cooldown release failed:', releaseErr?.message || releaseErr);
|
||||
}
|
||||
return Response.json({ error: 'storage_failed' }, { status: 500 });
|
||||
}
|
||||
|
||||
this.#broadcastPixels(pixels);
|
||||
@@ -116,8 +126,10 @@ export class CanvasRoom {
|
||||
if (!wasClean) {
|
||||
console.warn(`WS unclean close: code=${code} reason=${reason || '<none>'}`);
|
||||
}
|
||||
// Required pre-2026-04-07 compat date; harmless after.
|
||||
ws.close(code, reason);
|
||||
// Required because compatibility_date 2025-04-01 predates the
|
||||
// 2026-04-07 default-close cutoff. Remove if/when wrangler.json
|
||||
// bumps past that date. The try/catch handles already-closed sockets.
|
||||
try { ws.close(code, reason); } catch { /* already closed */ }
|
||||
}
|
||||
|
||||
webSocketError(ws, error) {
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { CANVAS_WIDTH, TOTAL_PIXELS, CHUNK_BYTES } from '../../lib/constants.js';
|
||||
import { CANVAS_WIDTH, TOTAL_PIXELS, CHUNK_BYTES, CHUNK_COUNT } from '../../lib/constants.js';
|
||||
|
||||
/**
|
||||
* Canvas pixel storage as fixed-size BLOB chunks in DO SQLite.
|
||||
@@ -42,23 +42,36 @@ export function readChunk(sql, chunkId) {
|
||||
/**
|
||||
* Read all chunks concatenated into a single TOTAL_PIXELS-sized buffer.
|
||||
* Used to serve GET /api/canvas.
|
||||
*
|
||||
* Bounded by chunk_id < CHUNK_COUNT so orphan rows left after a canvas-shrink
|
||||
* don't trip a RangeError on out.set(...).
|
||||
*/
|
||||
export function readAllChunks(sql) {
|
||||
const out = new Uint8Array(TOTAL_PIXELS);
|
||||
const cursor = sql.exec('SELECT chunk_id, bytes FROM canvas_chunks');
|
||||
const cursor = sql.exec(
|
||||
'SELECT chunk_id, bytes FROM canvas_chunks WHERE chunk_id < ?',
|
||||
CHUNK_COUNT,
|
||||
);
|
||||
for (const row of cursor) {
|
||||
const chunkId = row.chunk_id;
|
||||
const blob = row.bytes;
|
||||
const view = blob instanceof Uint8Array ? blob : new Uint8Array(blob);
|
||||
out.set(view, chunkId * CHUNK_BYTES);
|
||||
// Defensive clamp: if a row's persisted blob is longer than this chunk's
|
||||
// expected size (e.g. legacy data), trim before copying.
|
||||
const expected = chunkSize(chunkId);
|
||||
const trimmed = view.length > expected ? view.subarray(0, expected) : view;
|
||||
out.set(trimmed, chunkId * CHUNK_BYTES);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Apply a batch of pixel writes. Groups by chunk so each touched chunk
|
||||
* incurs at most one read and one write. Single transaction so the batch
|
||||
* is atomic across all touched chunks.
|
||||
* incurs at most one read and one write.
|
||||
*
|
||||
* Atomicity: each sql.exec auto-commits, so this function is NOT atomic on
|
||||
* its own. The caller must wrap the call in state.storage.transactionSync
|
||||
* (or transaction) to make a multi-chunk batch all-or-nothing.
|
||||
*
|
||||
* @param {SqlStorage} sql
|
||||
* @param {Array<{x:number, y:number, color:number}>} pixels
|
||||
@@ -79,17 +92,13 @@ export function writePixels(sql, pixels) {
|
||||
bucket.push({ byteOffset, color: p.color });
|
||||
}
|
||||
|
||||
// For each touched chunk: read current bytes (or zero-fill), apply edits,
|
||||
// write back. INSERT OR REPLACE upserts the row.
|
||||
// Atomicity: this loop is fully synchronous (no `await`) and the DO is
|
||||
// single-threaded, so the chunk updates are atomic with respect to other
|
||||
// requests. If anyone adds an `await` inside this loop, wrap the whole
|
||||
// block in `state.storage.transactionSync(() => { ... })` to preserve it.
|
||||
for (const [chunkId, edits] of groups) {
|
||||
const buf = readChunk(sql, chunkId);
|
||||
// readChunk returns a fresh Uint8Array (or wraps a buffer); writes must
|
||||
// not alias persisted state, so copy to be safe.
|
||||
const next = new Uint8Array(buf);
|
||||
// Allocate against chunkSize, never the persisted blob's length: after a
|
||||
// canvas grow, the old short blob would silently drop OOB writes.
|
||||
const expected = chunkSize(chunkId);
|
||||
const next = new Uint8Array(expected);
|
||||
next.set(buf.subarray(0, Math.min(buf.length, expected)));
|
||||
for (const { byteOffset, color } of edits) {
|
||||
next[byteOffset] = color;
|
||||
}
|
||||
|
||||
@@ -42,11 +42,13 @@ export function tryAcquire(sql, userId, now = Date.now()) {
|
||||
// No expired row to update. Either the user has never been seen (insert
|
||||
// succeeds) or they hold an active claim (insert fails on PK conflict).
|
||||
try {
|
||||
sql.exec(
|
||||
const insertCursor = sql.exec(
|
||||
'INSERT INTO cooldowns (user_id, expires_at) VALUES (?, ?)',
|
||||
userId,
|
||||
expiresAt,
|
||||
);
|
||||
// Drain symmetrically with the UPDATE branch so statement effects commit.
|
||||
insertCursor.toArray();
|
||||
if (Math.random() < GC_SAMPLE_RATE) {
|
||||
try { gc(sql, now); } catch { /* GC is best-effort */ }
|
||||
}
|
||||
@@ -56,6 +58,14 @@ export function tryAcquire(sql, userId, now = Date.now()) {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Refund a cooldown row so a transient write failure doesn't soft-DOS the
|
||||
* user for 1s. Called from the catch path in placePixels.
|
||||
*/
|
||||
export function release(sql, userId) {
|
||||
sql.exec('DELETE FROM cooldowns WHERE user_id = ?', userId).toArray();
|
||||
}
|
||||
|
||||
/** Delete all expired cooldown rows. Cheap with the expires_at index. */
|
||||
export function gc(sql, now = Date.now()) {
|
||||
sql.exec('DELETE FROM cooldowns WHERE expires_at <= ?', now).toArray();
|
||||
|
||||
@@ -18,6 +18,12 @@ export const MAX_BATCH_SIZE = 2048;
|
||||
export const CHUNK_BYTES = 65536;
|
||||
export const CHUNK_COUNT = Math.ceil(TOTAL_PIXELS / CHUNK_BYTES);
|
||||
|
||||
// CF DO SQLite has a per-cell BLOB cap (~2 MB at compat date 2025-04-01).
|
||||
// Fail-fast at module load if a future bump would overflow.
|
||||
if (CHUNK_BYTES > 2_000_000) {
|
||||
throw new Error(`CHUNK_BYTES (${CHUNK_BYTES}) exceeds DO SQLite per-cell BLOB cap (~2 MB)`);
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Build the 256-color palette deterministically:
|
||||
|
||||
+7
-1
@@ -21,7 +21,13 @@ app.get('/api/canvas', async (c) => {
|
||||
|
||||
/** POST /api/place — validate at the edge, forward to the DO. */
|
||||
app.post('/api/place', async (c) => {
|
||||
const contentLength = parseInt(c.req.header('content-length') || '0', 10);
|
||||
// Require a positive Content-Length. Missing or zero would otherwise let a
|
||||
// chunked-transfer-encoded body bypass the MAX_BODY_BYTES pre-parse cap.
|
||||
const contentLengthRaw = c.req.header('content-length');
|
||||
const contentLength = parseInt(contentLengthRaw ?? '', 10);
|
||||
if (!Number.isFinite(contentLength) || contentLength <= 0) {
|
||||
return c.json({ error: 'content_length_required' }, 411);
|
||||
}
|
||||
if (contentLength > MAX_BODY_BYTES) {
|
||||
return c.json({ error: 'body_too_large', max: MAX_BODY_BYTES }, 413);
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user