You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by as...@apache.org on 2026/08/24 17:40:17 UTC
(maka) 04/05: fix(storage): preserve transcript chunks during rebinding
This is an automated email from the ASF dual-hosted git repository.
Astro-Han pushed a commit to branch fix/durable-message-lifecycle
in repository https://gitbox.apache.org/repos/asf/maka.git
commit a8eabb1fab3b2a0091bde9233242ff17e4573bbe
Author: AstroHan <le...@outlook.com>
AuthorDate: Tue Aug 25 01:38:33 2026 +0800
fix(storage): preserve transcript chunks during rebinding
Generated-by: Codex
---
.../storage/src/sqlite-session-metadata-store.ts | 68 ++++++++++++++++++----
1 file changed, 58 insertions(+), 10 deletions(-)
diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts
index d2370a01d..d66c948d8 100644
--- a/packages/storage/src/sqlite-session-metadata-store.ts
+++ b/packages/storage/src/sqlite-session-metadata-store.ts
@@ -1826,11 +1826,7 @@ export class SqliteSessionMetadataStore {
throw new SessionMetadataConflictError('Message admission transcript Turn conflict');
}
const rebound = decodeCanonicalMessage({ ...message, turnId: input.turnId });
- this.db
- .prepare(
- 'UPDATE session_messages SET record_json = ? WHERE session_id = ? AND sequence = ?',
- )
- .run(JSON.stringify(rebound), input.sessionId, sequence);
+ this.replaceSessionMessageSync(input.sessionId, sequence, rebound);
}
});
}
@@ -1922,11 +1918,7 @@ export class SqliteSessionMetadataStore {
if (typeof sequence !== 'number' || !Number.isSafeInteger(sequence)) {
throw new SessionMetadataConflictError('Invalid Message transcript sequence');
}
- this.db
- .prepare(
- 'UPDATE session_messages SET record_json = ? WHERE session_id = ? AND sequence = ?',
- )
- .run(json, stored.sessionId, sequence);
+ this.replaceSessionMessageSync(stored.sessionId, sequence, message, json);
}
this.updateCatalogProjectionSync(
stored.sessionId,
@@ -4765,6 +4757,62 @@ export class SqliteSessionMetadataStore {
}
}
+ private replaceSessionMessageSync(
+ sessionId: string,
+ sequence: number,
+ message: StoredMessage,
+ json = JSON.stringify(message),
+ ): void {
+ const encoded = Buffer.from(json, 'utf8');
+ this.db
+ .prepare('DELETE FROM session_message_chunks WHERE session_id = ? AND sequence = ?')
+ .run(sessionId, sequence);
+ this.db
+ .prepare('DELETE FROM session_message_payloads WHERE session_id = ? AND sequence = ?')
+ .run(sessionId, sequence);
+ if (encoded.byteLength <= SQLITE_SESSION_MESSAGE_CHUNK_BYTES) {
+ this.db
+ .prepare(
+ 'UPDATE session_messages SET record_json = ? WHERE session_id = ? AND sequence = ?',
+ )
+ .run(json, sessionId, sequence);
+ return;
+ }
+ this.db
+ .prepare(
+ 'UPDATE session_messages SET record_json = ? WHERE session_id = ? AND sequence = ?',
+ )
+ .run(SQLITE_SESSION_MESSAGE_CHUNK_MARKER, sessionId, sequence);
+ this.db
+ .prepare(
+ 'INSERT INTO session_message_payloads(session_id, sequence, record_bytes, sha256) VALUES (?, ?, ?, ?)',
+ )
+ .run(
+ sessionId,
+ sequence,
+ encoded.byteLength,
+ createHash('sha256').update(encoded).digest('hex'),
+ );
+ for (
+ let offset = 0;
+ offset < encoded.byteLength;
+ offset += SQLITE_SESSION_MESSAGE_CHUNK_BYTES
+ ) {
+ const chunk = encoded.subarray(offset, offset + SQLITE_SESSION_MESSAGE_CHUNK_BYTES);
+ this.db
+ .prepare(
+ 'INSERT INTO session_message_chunks(session_id, sequence, chunk_index, data, sha256) VALUES (?, ?, ?, ?, ?)',
+ )
+ .run(
+ sessionId,
+ sequence,
+ offset / SQLITE_SESSION_MESSAGE_CHUNK_BYTES,
+ chunk,
+ createHash('sha256').update(chunk).digest('hex'),
+ );
+ }
+ }
+
private readMessagesWith(
sessionId: string,
decode: (value: unknown) => StoredMessage,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org