chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
fix(repo): refuse a create at a key the repo already holds
.changeset/create-refuses-existing.mdnew+11−0
| @@ -0,0 +1,11 @@ | ||
--- | ||
'@pdsjs/core': patch | ||
--- | ||
| ||
A create at a key the repo already holds is refused with 400 | ||
`InvalidRequest` ("Record already exists"), as the reference PDS does, | ||
instead of replacing the record. That covers `createRecord` and | ||
`applyWrites#create`, including two creates of one key in one batch, and | ||
a refused batch writes nothing. Clients claim a key by creating at it: | ||
knotless names a repository's record by its name and relies on the create | ||
failing when the name is taken. | ||
packages/core/src/handlers/xrpc-repo.js+27−14
| @@ -16,6 +16,9 @@ import { | ||
} from '../repo.js'; | ||
import { ScopePermissions } from '../scope.js'; | ||
| ||
/** What createRecord's locked write returns when the key is taken. */ | ||
const EXISTS = Symbol('exists'); | ||
| ||
/** | ||
* @typedef {Object} RepoContext | ||
* @property {import('../ports.js').ActorStoragePort} actorStorage | ||
| @@ -52,6 +55,18 @@ export function createRepoHandlers(ctx) { | ||
writeGuard, | ||
} = ctx; | ||
| ||
/** | ||
* The answer to a create at a key the repo already holds. | ||
* @param {string} uri | ||
* @returns {Response} | ||
*/ | ||
function recordExists(uri) { | ||
return Response.json( | ||
{ error: 'InvalidRequest', message: `Record already exists: ${uri}` }, | ||
{ status: 400 }, | ||
); | ||
} | ||
| ||
/** | ||
* Consult the write guard for one write in a collection it covers. Null to | ||
* proceed, the refusal Response otherwise. | ||
| @@ -221,11 +236,8 @@ export function createRepoHandlers(ctx) { | ||
if (refused) return refused; | ||
| ||
const latestCommit = await withRepoLock(async () => { | ||
// A create whose key the repo already holds overwrites that record, so | ||
// the tree performs an update. The op announced has to be the op the | ||
// tree ran: calling it a create puts an op on the firehose that the | ||
// diff does not contain, and the commit cannot be proven. | ||
const existing = await actorStorage.getRecord(uri); | ||
// Clients claim a key by creating at it, so a create never replaces. | ||
if (await actorStorage.getRecord(uri)) return EXISTS; | ||
| ||
await actorStorage.putBlock(cid, encoded); | ||
await actorStorage.putRecord(uri, cid, collection, rkey, encoded); | ||
| @@ -235,11 +247,10 @@ export function createRepoHandlers(ctx) { | ||
await actorStorage.linkBlobToRecord(blobCid, uri, recordTimeOf(record)); | ||
} | ||
| ||
await createCommit(did, [ | ||
{ action: existing ? 'update' : 'create', uri, cid }, | ||
]); | ||
await createCommit(did, [{ action: 'create', uri, cid }]); | ||
return actorStorage.getLatestCommit(); | ||
}); | ||
if (latestCommit === EXISTS) return recordExists(uri); | ||
| ||
return Response.json({ | ||
uri, | ||
| @@ -592,6 +603,8 @@ export function createRepoHandlers(ctx) { | ||
// The write guard, before anything lands, for the same reason as the | ||
// scope pre-pass. A create with no rkey mints one below; a minted TID | ||
// never names a config record, so the guard sees the empty rkey. | ||
// The keys this batch creates, so two creates of one key are refused. | ||
const claimed = new Set(); | ||
for (const write of writes) { | ||
const action = String(write?.$type ?? '').split('#')[1]; | ||
if (action !== 'create' && action !== 'update' && action !== 'delete') { | ||
| @@ -603,6 +616,11 @@ export function createRepoHandlers(ctx) { | ||
`at://${did}/${write.collection}/${rkey}`, | ||
) | ||
: null; | ||
const uri = `at://${did}/${write.collection}/${rkey}`; | ||
if (action === 'create' && rkey) { | ||
if (existing || claimed.has(uri)) return recordExists(uri); | ||
claimed.add(uri); | ||
} | ||
const refused = await guardWrite( | ||
write.collection, | ||
rkey, | ||
| @@ -621,11 +639,6 @@ export function createRepoHandlers(ctx) { | ||
const rkey = providedRkey || createTid(); | ||
const uri = `at://${did}/${collection}/${rkey}`; | ||
| ||
// A create over a key the repo already holds is an update as far as | ||
// the tree is concerned; see createRecord. The announced action | ||
// follows the tree, not the caller's wording. | ||
const present = await actorStorage.getRecord(uri); | ||
| ||
// Ensure $type is set | ||
const recordWithType = { $type: collection, ...value }; | ||
| ||
| @@ -673,7 +686,7 @@ export function createRepoHandlers(ctx) { | ||
); | ||
} | ||
| ||
ops.push({ action: present ? 'update' : 'create', uri, cid }); | ||
ops.push({ action: 'create', uri, cid }); | ||
results.push({ | ||
$type: 'com.atproto.repo.applyWrites#createResult', | ||
uri, | ||
packages/core/test/concurrent-writes.test.js+5−8
| @@ -168,20 +168,17 @@ describe('concurrent record writes', () => { | ||
expect((await storage.listAllRecords()).length).toBe(9); | ||
}); | ||
| ||
test('concurrent creates at one key commit one create and then updates', async () => { | ||
test('concurrent creates at one key commit one create and refuse the rest', async () => { | ||
const results = await Promise.all( | ||
['a', 'b', 'c', 'd'].map((status) => | ||
write('createRecord', { rkey: 'same', record: check(status) }), | ||
), | ||
); | ||
for (const res of results) expect(res.status).toBe(200); | ||
expect(results.map((res) => res.status).sort()).toEqual([ | ||
200, 400, 400, 400, | ||
]); | ||
| ||
const chain = await expectContinuousChain(); | ||
expect(chain.map((c) => c.evt.ops[0].action)).toEqual([ | ||
'create', | ||
'update', | ||
'update', | ||
'update', | ||
]); | ||
expect(chain.map((c) => c.evt.ops[0].action)).toEqual(['create']); | ||
}); | ||
}); | ||
test/xrpc-repo.test.js+90−0
| @@ -235,6 +235,30 @@ describe('createRecord', () => { | ||
expect(body.uri).toBe(`at://${DID}/${COLLECTION}/chosen`); | ||
}); | ||
| ||
it('refuses a key the repo already holds, and keeps the record', async () => { | ||
const { repo, records, commits } = createRepo(); | ||
const body = { repo: DID, collection: COLLECTION, rkey: 'taken' }; | ||
const first = await ( | ||
await call(repo, CREATE, { | ||
auth: FULL, | ||
body: { ...body, record: { text: 'first' } }, | ||
}) | ||
).json(); | ||
| ||
const response = await call(repo, CREATE, { | ||
auth: FULL, | ||
body: { ...body, record: { text: 'second' } }, | ||
}); | ||
| ||
expect(response.status).toBe(400); | ||
expect(await response.json()).toEqual({ | ||
error: 'InvalidRequest', | ||
message: `Record already exists: ${first.uri}`, | ||
}); | ||
expect(records.get(first.uri)?.cid).toBe(first.cid); | ||
expect(commits).toHaveLength(1); | ||
}); | ||
| ||
it('needs a session', async () => { | ||
const { repo } = createRepo(); | ||
const response = await call(repo, CREATE, { | ||
| @@ -624,6 +648,72 @@ describe('applyWrites', () => { | ||
expect(commits[0]).toHaveLength(2); | ||
}); | ||
| ||
it('refuses a create at a key the repo already holds, writing nothing', async () => { | ||
const { repo, records, commits } = createRepo(); | ||
const first = await ( | ||
await call(repo, CREATE, { | ||
auth: FULL, | ||
body: { | ||
repo: DID, | ||
collection: COLLECTION, | ||
rkey: 'taken', | ||
record: { text: 'first' }, | ||
}, | ||
}) | ||
).json(); | ||
| ||
const response = await call(repo, APPLY, { | ||
auth: FULL, | ||
body: { | ||
repo: DID, | ||
writes: [ | ||
{ | ||
$type: 'com.atproto.repo.applyWrites#create', | ||
collection: COLLECTION, | ||
rkey: 'other', | ||
value: { text: 'other' }, | ||
}, | ||
{ | ||
$type: 'com.atproto.repo.applyWrites#create', | ||
collection: COLLECTION, | ||
rkey: 'taken', | ||
value: { text: 'second' }, | ||
}, | ||
], | ||
}, | ||
}); | ||
| ||
expect(response.status).toBe(400); | ||
expect((await response.json()).message).toBe( | ||
`Record already exists: ${first.uri}`, | ||
); | ||
expect(records.size).toBe(1); | ||
expect(records.get(first.uri)?.cid).toBe(first.cid); | ||
expect(commits).toHaveLength(1); | ||
}); | ||
| ||
it('refuses two creates of one key in a batch', async () => { | ||
const { repo, records, commits } = createRepo(); | ||
const create = { | ||
$type: 'com.atproto.repo.applyWrites#create', | ||
collection: COLLECTION, | ||
rkey: 'twice', | ||
value: { text: 'a' }, | ||
}; | ||
| ||
const response = await call(repo, APPLY, { | ||
auth: FULL, | ||
body: { | ||
repo: DID, | ||
writes: [create, { ...create, value: { text: 'b' } }], | ||
}, | ||
}); | ||
| ||
expect(response.status).toBe(400); | ||
expect(records.size).toBe(0); | ||
expect(commits).toHaveLength(0); | ||
}); | ||
| ||
it('refuses a read-only server', async () => { | ||
const { repo } = createRepo({ readOnly: true }); | ||
const response = await call(repo, APPLY, { | ||