Nothing enters the catalog without a usable name: the pending store (decision f amended)
An unnamed release (no readable file name) is written to pending_releases (migration 6300, the HDD) with its keys, counts, article span, upload tool and probe targets, instead of releases, release_texts, posters and a chunk. The edge sends it with an empty payload and its tool; the batch store splits a batch by readable. The pending ladder (probes cheapest first, signature verdicts, the move into the catalog) comes next. The replay report shows the pending releases. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
1 parent
6860e6ccaf
commit
9cf85e09d8
8 files changed
+286
-15
No files matched your search
@@ -8,6 +8,7 @@ import { decodeChunk } from '#lib/chunks/index'
|
||||
import { SIDECAR } from '#services/sidecar_folder'
|
||||
import { identityKey, releaseKey } from '#lib/release/keys'
|
||||
import { sourceCode } from '#lib/release/source'
|
||||
import { uploadTool } from '#lib/rules/index'
|
||||
import { TEXT_KINDS, TEXT_MAX, textsToStore } from '#lib/release/text_kind'
|
||||
|
||||
/**
|
||||
@@ -30,6 +31,8 @@ export interface StoredBatch {
|
||||
/** Provisional (tail) releases the batch finished. */
|
||||
finalized: number
|
||||
candidates: number
|
||||
/** Unnamed releases written to the pending store instead of the catalog (decision f). */
|
||||
pending: number
|
||||
}
|
||||
|
||||
/** A chunk's slot in its batch: the key and the upload the edge numbered it with (ProbeCandidate points at it). */
|
||||
@@ -73,18 +76,29 @@ class BatchStore {
|
||||
newReleases: 0,
|
||||
finalized: 0,
|
||||
candidates: 0,
|
||||
pending: 0,
|
||||
}
|
||||
|
||||
// Nothing enters the catalog without a usable name (decision f): an unnamed release (no readable file name)
|
||||
// goes to the pending store with its probe targets, the named ones to the catalog.
|
||||
const named = batch.releases.filter((r) => r.readable)
|
||||
const unnamed = batch.releases.filter((r) => !r.readable)
|
||||
const written = await this.writeReleases(
|
||||
trx,
|
||||
{ groupId: Number(range.group_id), blockId },
|
||||
batch.from,
|
||||
batch.releases,
|
||||
named,
|
||||
batch.candidates
|
||||
)
|
||||
result.newReleases = written.newReleases
|
||||
result.finalized = written.finalized
|
||||
result.candidates = written.candidates
|
||||
result.pending = await this.writePending(
|
||||
trx,
|
||||
Number(range.group_id),
|
||||
unnamed,
|
||||
batch.candidates
|
||||
)
|
||||
|
||||
if (batch.counts.length > 0) {
|
||||
await trx.rawQuery(
|
||||
@@ -153,6 +167,68 @@ class BatchStore {
|
||||
return Number(block.id)
|
||||
}
|
||||
|
||||
/**
|
||||
* The unnamed releases of a batch into pending_releases (migration 6300, the HDD): keys, counts, the upload
|
||||
* tool and the probe targets, no chunk. The pending ladder (REBUILD.md decision f) probes them cheapest first,
|
||||
* and a name moves one into the catalog; a provisional tail release sent again adds its counts up.
|
||||
*/
|
||||
async writePending(
|
||||
trx: TransactionClientContract,
|
||||
groupId: number,
|
||||
releases: ReleaseChunk[],
|
||||
candidates: ProbeCandidate[]
|
||||
): Promise<number> {
|
||||
if (releases.length === 0) return 0
|
||||
const targets = new Map<string, { kind: string; messageId: string; file: string }[]>()
|
||||
for (const c of candidates) {
|
||||
const key = slot(c.releaseKey, c.upload ?? 0)
|
||||
const list = targets.get(key) ?? []
|
||||
list.push({ kind: c.kind, messageId: c.messageId, file: c.file })
|
||||
targets.set(key, list)
|
||||
}
|
||||
let written = 0
|
||||
for (let i = 0; i < releases.length; i += 1000) {
|
||||
const rows = releases.slice(i, i + 1000).map((r) => {
|
||||
const probes = targets.get(slot(r.key, r.upload)) ?? []
|
||||
return {
|
||||
release_key: releaseKey(r.key, groupId, r.firstArticle).toString('hex'),
|
||||
identity_key: identityKey(r.key).toString('hex'),
|
||||
name: r.name.slice(0, TEXT_MAX),
|
||||
poster: r.poster || null,
|
||||
first: r.postedAt,
|
||||
last: r.postedLast ?? r.postedAt,
|
||||
first_article: r.firstArticle,
|
||||
last_article: r.lastArticle,
|
||||
files: r.files,
|
||||
segments: r.segments,
|
||||
bytes: r.bytes,
|
||||
partial: r.partial === true,
|
||||
finished: r.finished !== false,
|
||||
tool: r.tool ?? (probes[0] ? uploadTool(probes[0].messageId) : 'unknown'),
|
||||
candidates: probes,
|
||||
}
|
||||
})
|
||||
const { rows: done } = await trx.rawQuery(
|
||||
`INSERT INTO pending_releases (release_key, identity_key, group_id, name, poster, posted_at, last_posted_at, first_article, last_article,
|
||||
files, segments, bytes, partial, finished, tool, candidates)
|
||||
SELECT decode(v.release_key, 'hex'), decode(v.identity_key, 'hex'), ?, v.name, v.poster, to_timestamp(v.first / 1000.0), to_timestamp(v.last / 1000.0),
|
||||
v.first_article, v.last_article, v.files, v.segments, v.bytes, v.partial, v.finished, v.tool, v.candidates
|
||||
FROM jsonb_to_recordset(?::jsonb) AS v(release_key text, identity_key text, name text, poster text, first bigint, last bigint, first_article bigint, last_article bigint,
|
||||
files int, segments bigint, bytes bigint, partial boolean, finished boolean, tool text, candidates jsonb)
|
||||
ON CONFLICT (release_key) DO UPDATE SET
|
||||
files = pending_releases.files + excluded.files, segments = pending_releases.segments + excluded.segments, bytes = pending_releases.bytes + excluded.bytes,
|
||||
last_posted_at = greatest(pending_releases.last_posted_at, excluded.last_posted_at), last_article = greatest(pending_releases.last_article, excluded.last_article),
|
||||
partial = excluded.partial, finished = excluded.finished,
|
||||
candidates = CASE WHEN jsonb_array_length(pending_releases.candidates) = 0 THEN excluded.candidates ELSE pending_releases.candidates END
|
||||
WHERE NOT pending_releases.finished
|
||||
RETURNING release_key`,
|
||||
[groupId, JSON.stringify(rows)]
|
||||
)
|
||||
written += done.length
|
||||
}
|
||||
return written
|
||||
}
|
||||
|
||||
/**
|
||||
* The releases of a batch into the catalog, one chunk each, and their probe candidates. Every chunk is one
|
||||
* release: inserted, or (a provisional tail release sent again) finished.
|
||||
|
||||
Reference in new issue
Block a user