perf(logdb): incremental append-only sealing for immediate mode
ober
5869f4202245bb27ee8319f8cb925f3bae26d2ea
--- a/Makefile +++ b/Makefile @@ -93,6 +93,7 @@ test: binary native-test-stage $(JEXEC) tests/test-bounded-line.ss $(JEXEC) tests/test-logdb-jsqlite.ss $(JEXEC) tests/test-logdb-crypto.ss + $(JEXEC) tests/test-logdb-incremental.ss $(JEXEC) tests/test-secret-input.ss $(JEXEC) tests/test-attachments.ss $(JEXEC) tests/test-native-loader.ss --- a/signal/log_crypto.ss +++ b/signal/log_crypto.ss @@ -6,7 +6,8 @@ log-random-bytes log-scrypt-key log-aead-seal - log-aead-open) + log-aead-open + log-sha256) (import (except (scheme) make-hash-table hash-table? @@ -154,4 +155,16 @@ (error 'log-aead-open "AEAD open failed" (last-error))) (bv-sub out 0 (result-length len-buf)))) + (def (log-sha256 input) + (unless (entry? "jerboa_sha256") + (error 'log-sha256 "native SHA-256 is not available")) + (let* ([in (as-bytes input)] + [out (make-bytevector 32 0)] + [proc (foreign-procedure "jerboa_sha256" + (u8* size_t u8* size_t) int)] + [rc (proc in (bytevector-length in) out 32)]) + (unless (= rc 32) + (error 'log-sha256 "SHA-256 failed" (last-error))) + out)) + ) ;; end library --- a/signal/logdb.ss +++ b/signal/logdb.ss @@ -10,7 +10,9 @@ logdb-open logdb-close logdb-put logdb-count logdb-recent logdb-migrate-legacy-to-jsqlite logdb-prompt-passphrase - logdb-handle-key-zeroed?) + logdb-handle-key-zeroed? + logdb-handle-full-seal-count + logdb-handle-delta-frame-count) (import (except (scheme) make-hash-table hash-table? @@ -40,7 +42,9 @@ (defstruct jlog (path key salt generation db dirty? tx-open? row-count payload-bytes pending-rows pending-bytes last-persist-ms - lock closed? flush-thread flush-stop-box)) + lock closed? flush-thread flush-stop-box + incremental? pending-data last-seal-hash delta-frames + ckpt-generation full-seal-count pruned-since-seal?)) ;; jsqlite query/exec wrappers expose no retained prepared-statement handle. (def close-jsqlite-database sqlite-close) @@ -370,6 +374,252 @@ (write-file-atomic! (generation-marker-path path) (generation->bytes generation))) + ;; -------------------------------------------------------------------------- + ;; Incremental append-only sealing (immediate mode). + ;; + ;; A SQLite image (sqlite-db->bytevector) is NOT append-only -- inserting one + ;; row can rewrite arbitrary pages (B-tree rebalancing, header counters, the + ;; freelist) -- so it cannot be sealed as a byte delta of the previous image. + ;; Immediate mode therefore persists an append-only chain of sealed DELTA + ;; frames (the new rows only) instead of re-sealing the whole db: + ;; + ;; path small "head" container, resealed per put in O(1): + ;; magic || salt || generation || AEAD(head-state). Its + ;; generation advances on every persist (the counter nonce the + ;; crypto regressions assert) without serializing the whole db. + ;; path.ckpt whole-db checkpoint container (legacy single-container + ;; format): magic || salt || gen || AEAD(db). Rewritten only at a + ;; checkpoint (creation, periodic, prune, batch/close modes). + ;; path.delta append-only sealed delta frames. Frame layout: + ;; "JDLT" || generation(8 BE) || sealed-len(8 BE) || AEAD(rows). + ;; + ;; Integrity argument: every frame is AEAD-sealed under the scrypt-derived key + ;; with a unique domain-tagged nonce (tag 3 || generation) and an AAD binding + ;; salt, generation, and the PREVIOUS chain hash, where + ;; chain-hash = SHA256(prev-hash || sealed-frame). Tampering breaks the AEAD + ;; tag; removing, reordering, or rolling back a frame breaks the hash link and + ;; the consecutive-generation sequence; the expected final chain hash is itself + ;; authenticated inside the head container, so the open path fails closed on + ;; any divergence. The external .gen marker still guards whole-container + ;; rollback and a wrong passphrase fails closed when the head is opened. + ;; Nonces are domain-tagged (head=1, frame=3; the checkpoint reuses the legacy + ;; tag-0 counter nonce) so head, checkpoint, and frames never reuse a + ;; (key, nonce) pair, even at the same generation. Batch/close modes keep the + ;; legacy single-container full reseal untouched. + ;; + ;; Residual: a prune rewrites arbitrary pages, so a put that prunes forces a + ;; full checkpoint instead of a delta frame; this keeps delta frames + ;; insert-only and replay trivial. + + (def *jlog-head-tag* 1) + (def *jlog-ckpt-tag* 2) + (def *jlog-frame-tag* 3) + (def *jlog-delta-magic* (string->utf8 "JDLT")) + (def *jlog-chain-hash-len* 32) + (def *default-log-checkpoint-every* 64) + (def *hard-log-checkpoint-every* 100000) + + (def (log-checkpoint-every) + (bounded-positive-env "JERBOA_SIGNAL_LOG_CHECKPOINT_EVERY" + *default-log-checkpoint-every* + *hard-log-checkpoint-every*)) + + (def (checkpoint-path path) (string-append path ".ckpt")) + (def (delta-path path) (string-append path ".delta")) + + (def (bv-equal? a b) + (and (= (bytevector-length a) (bytevector-length b)) + (let loop ([i 0]) + (or (= i (bytevector-length a)) + (and (= (bytevector-u8-ref a i) (bytevector-u8-ref b i)) + (loop (+ i 1))))))) + + ;; Domain-tagged nonce: tag byte || generation, so head/checkpoint/frame seals + ;; never share a (key, nonce) pair even at the same generation. + (def (domain-nonce tag generation) + (let ([nonce (make-bytevector *jlog-nonce-len* 0)]) + (bytevector-u8-set! nonce 0 tag) + (bytevector-u64-set! nonce (- *jlog-nonce-len* *jlog-generation-len*) + generation (endianness big)) + nonce)) + + (def (incremental-aad tag salt nonce generation-bytes prev-hash) + (bv-append *jlog-magic* (bytevector tag) salt nonce generation-bytes + prev-hash)) + + (def (seal-incremental key tag salt generation plain prev-hash) + (let* ([nonce (domain-nonce tag generation)] + [aad (incremental-aad tag salt nonce (generation->bytes generation) + prev-hash)]) + (log-aead-seal key nonce plain aad))) + + (def (open-incremental key tag salt generation sealed prev-hash) + (let* ([nonce (domain-nonce tag generation)] + [aad (incremental-aad tag salt nonce (generation->bytes generation) + prev-hash)]) + (guard (e [(condition? e) (raise-passphrase-auth-error)]) + (log-aead-open key nonce sealed aad)))) + + ;; Envelope for the head container: magic || salt || generation(8 BE) || sealed. + (def (pack-incremental-container salt generation sealed) + (bv-append *jlog-magic* salt (generation->bytes generation) sealed)) + + (def (unpack-incremental-container bytes) + (let* ([magic-len (bytevector-length *jlog-magic*)] + [need (+ magic-len *jlog-salt-len* *jlog-generation-len*)]) + (when (< (bytevector-length bytes) need) + (error 'logdb-open "encrypted jsqlite log is too short")) + (let* ([salt (bv-sub bytes magic-len *jlog-salt-len*)] + [generation (bytes->generation + (bv-sub bytes (+ magic-len *jlog-salt-len*) + *jlog-generation-len*))] + [sealed (bv-sub bytes need (- (bytevector-length bytes) need))]) + (values salt generation sealed)))) + + (def (checkpoint-chain-hash ckpt-container-bytes) + (log-sha256 ckpt-container-bytes)) + + ;; head-state: ckpt-generation(u64) || delta-count(u64) || row-count(u64) || + ;; chain-hash(32). + (def *head-state-len* (+ 8 8 8 *jlog-chain-hash-len*)) + + (def (pack-head-state ckpt-generation delta-count row-count chain-hash) + (let ([bv (make-bytevector *head-state-len* 0)]) + (bytevector-u64-set! bv 0 ckpt-generation (endianness big)) + (bytevector-u64-set! bv 8 delta-count (endianness big)) + (bytevector-u64-set! bv 16 row-count (endianness big)) + (bytevector-copy! chain-hash 0 bv 24 *jlog-chain-hash-len*) + bv)) + + (def (unpack-head-state bv) + (unless (>= (bytevector-length bv) *head-state-len*) + (error 'logdb-open "incremental log head state is too short")) + (values (bytevector-u64-ref bv 0 (endianness big)) + (bytevector-u64-ref bv 8 (endianness big)) + (bytevector-u64-ref bv 16 (endianness big)) + (bv-sub bv 24 *jlog-chain-hash-len*))) + + ;; Delta rows are serialized insert-only: + ;; row = logged_at(u64) || timestamp(u64) || 7 * (len(u32) || utf8) + ;; fields: account direction conversation sender kind body raw + ;; rows = count(u64) || * (row-len(u32) || row) + (def (row->bytes row) + (let* ([strs (map (lambda (s) (string->utf8 (if (string? s) s ""))) + (list (list-ref row 1) (list-ref row 2) (list-ref row 3) + (list-ref row 4) (list-ref row 6) (list-ref row 7) + (list-ref row 8)))] + [total (+ 16 (let loop ([xs strs] [acc 0]) + (if (null? xs) acc + (loop (cdr xs) + (+ acc 4 (bytevector-length (car xs)))))))] + [bv (make-bytevector total 0)]) + (bytevector-u64-set! bv 0 (list-ref row 0) (endianness big)) + (bytevector-u64-set! bv 8 (list-ref row 5) (endianness big)) + (let loop ([xs strs] [off 16]) + (unless (null? xs) + (let ([b (car xs)]) + (bytevector-u32-set! bv off (bytevector-length b) (endianness big)) + (bytevector-copy! b 0 bv (+ off 4) (bytevector-length b)) + (loop (cdr xs) (+ off 4 (bytevector-length b)))))) + bv)) + + (def (parse-row bv) + (let ([logged-at (bytevector-u64-ref bv 0 (endianness big))] + [timestamp (bytevector-u64-ref bv 8 (endianness big))]) + (let loop ([off 16] [n 0] [acc '()]) + (if (= n 7) + (let ([s (reverse acc)]) + (list logged-at (list-ref s 0) (list-ref s 1) (list-ref s 2) + (list-ref s 3) timestamp (list-ref s 4) (list-ref s 5) + (list-ref s 6))) + (let* ([len (bytevector-u32-ref bv off (endianness big))] + [str (utf8->string (bv-sub bv (+ off 4) len))]) + (loop (+ off 4 len) (+ n 1) (cons str acc))))))) + + (def (rows->bytes rows) + (let* ([row-bvs (map row->bytes rows)] + [total (+ 8 (let loop ([xs row-bvs] [acc 0]) + (if (null? xs) acc + (loop (cdr xs) + (+ acc 4 (bytevector-length (car xs)))))))] + [bv (make-bytevector total 0)]) + (bytevector-u64-set! bv 0 (length rows) (endianness big)) + (let loop ([xs row-bvs] [off 8]) + (unless (null? xs) + (let ([b (car xs)]) + (bytevector-u32-set! bv off (bytevector-length b) (endianness big)) + (bytevector-copy! b 0 bv (+ off 4) (bytevector-length b)) + (loop (cdr xs) (+ off 4 (bytevector-length b)))))) + bv)) + + (def (bytes->rows bv) + (let ([count (bytevector-u64-ref bv 0 (endianness big))]) + (let loop ([off 8] [n 0] [acc '()]) + (if (= n count) + (reverse acc) + (let* ([len (bytevector-u32-ref bv off (endianness big))] + [row (parse-row (bv-sub bv (+ off 4) len))]) + (loop (+ off 4 len) (+ n 1) (cons row acc))))))) + + (def (append-delta-frame! path generation sealed) + (let* ([safe-path (checked-log-path (delta-path path))] + [mlen (bytevector-length *jlog-delta-magic*)] + [header (make-bytevector (+ mlen 8 8) 0)]) + (bytevector-copy! *jlog-delta-magic* 0 header 0 mlen) + (bytevector-u64-set! header mlen generation (endianness big)) + (bytevector-u64-set! header (+ mlen 8) (bytevector-length sealed) + (endianness big)) + (unless (file-exists? safe-path) + (write-file-bytevector! safe-path (make-bytevector 0 0))) + (let ([port (open-file-input/output-port safe-path + (file-options no-fail no-truncate) (buffer-mode none))]) + (guard (e [(condition? e) (close-port port) (raise e)]) + (set-port-position! port (port-length port)) + (put-bytevector port header) + (put-bytevector port sealed) + (close-port port))) + (chmod safe-path #o600))) + + ;; Parse complete frames; a torn trailing frame (incomplete header/body) is + ;; ignored, which is safe because the head's authenticated chain hash and the + ;; external generation marker detect any genuine loss. + (def (read-delta-frames path) + (let ([safe-path (delta-path path)]) + (if (not (file-exists? safe-path)) + '() + (let* ([bv (read-file-bytevector* safe-path)] + [mlen (bytevector-length *jlog-delta-magic*)] + [hlen (+ mlen 8 8)] + [total (bytevector-length bv)]) + (let loop ([off 0] [acc '()]) + (if (< (- total off) hlen) + (reverse acc) + (let* ([magic-ok (bv-prefix? *jlog-delta-magic* + (bv-sub bv off mlen))] + [generation (bytevector-u64-ref bv (+ off mlen) + (endianness big))] + [sealed-len (bytevector-u64-ref bv (+ off mlen 8) + (endianness big))] + [sealed-start (+ off hlen)] + [sealed-end (+ sealed-start sealed-len)]) + (if (or (not magic-ok) (> sealed-end total)) + (reverse acc) + (loop sealed-end + (cons (cons generation (bv-sub bv sealed-start sealed-len)) + acc)))))))))) + + (def (replay-insert-row! db row) + (sqlite-exec db *insert-sql* + (list-ref row 0) + (empty->null (list-ref row 1)) + (empty->null (list-ref row 2)) + (empty->null (list-ref row 3)) + (empty->null (list-ref row 4)) + (list-ref row 5) + (empty->null (list-ref row 6)) + (empty->null (list-ref row 7)) + (or (list-ref row 8) ""))) + (def (ensure-schema! db) (for-each (lambda (sql) (sqlite-exec db sql)) *schema-sql*)) @@ -377,11 +627,24 @@ (let* ([salt (log-random-bytes *jlog-salt-len*)] [key (derive-key passphrase salt)] [db (sqlite-open-bytevector (make-bytevector 0 0))] + [incremental? (persist-on-write?)] [log (make-jlog path key salt 0 db #f #f 0 0 0 0 (real-time) (make-mutex) #f - #f (box #f))]) + #f (box #f) + incremental? '() + (make-bytevector *jlog-chain-hash-len* 0) + 0 0 0 #f)]) (ensure-schema! db) - (persist-jlog! log) + ;; Drop any stale auxiliary files left by a previous log at this path. + (delete-if-exists! (checkpoint-path path)) + (delete-if-exists! (delta-path path)) + (if incremental? + (begin + (write-checkpoint! log 1) + (write-head! log 1 1 0 0 (jlog-last-seal-hash log)) + (write-generation-marker! path 1) + (jlog-generation-set! log 1)) + (persist-jlog! log)) (start-jlog-flusher! log) (make-logdb-handle 'jsqlite log))) @@ -400,10 +663,78 @@ (let ([log (make-jlog path key salt generation db #f #f rows payload 0 0 (real-time) (make-mutex) #f - #f (box #f))]) + #f (box #f) + #f '() + (make-bytevector *jlog-chain-hash-len* 0) + 0 0 0 #f)]) (start-jlog-flusher! log) (make-logdb-handle 'jsqlite log))))))) + ;; Open an incremental log: decrypt the head container (fails closed on a wrong + ;; passphrase), enforce the anti-rollback marker, load the checkpoint db, then + ;; replay and verify the sealed delta chain up to the authenticated head hash. + (def (open-incremental-jlog path passphrase bytes) + (let-values ([(salt head-generation head-sealed) + (unpack-incremental-container bytes)]) + (let* ([key (derive-key passphrase salt)] + [marker (read-generation-marker path)] + [head-state (open-incremental key *jlog-head-tag* salt + head-generation head-sealed + (make-bytevector 0 0))]) + (when (< head-generation marker) + (error 'logdb-open + "encrypted log generation regression detected (rollback)" + path head-generation marker)) + (write-generation-marker! path head-generation) + (let-values ([(ckpt-generation delta-count head-row-count chain-hash) + (unpack-head-state head-state)]) + (let ([ckpt-bytes (read-file-bytevector* (checkpoint-path path))]) + (unless (jlog-container-bytes? ckpt-bytes) + (error 'logdb-open "incremental log checkpoint missing" path)) + (let-values ([(ckpt-key ckpt-salt ckpt-gen ckpt-plain) + (decrypt-container ckpt-bytes passphrase)]) + (unless (= ckpt-gen ckpt-generation) + (error 'logdb-open + "incremental log checkpoint generation mismatch" + path ckpt-gen ckpt-generation)) + (let ([db (sqlite-open-bytevector ckpt-plain)]) + (ensure-schema! db) + (let replay ([xs (read-delta-frames path)] + [prev (checkpoint-chain-hash ckpt-bytes)] + [expected-gen (+ ckpt-gen 1)] + [count 0]) + (if (not (null? xs)) + (let* ([frame (car xs)] + [gen (car frame)] + [sealed (cdr frame)]) + (unless (= gen expected-gen) + (error 'logdb-open + "incremental log delta generation sequence broken" + path gen expected-gen)) + (let ([plain (open-incremental key *jlog-frame-tag* salt + gen sealed prev)]) + (for-each (lambda (row) (replay-insert-row! db row)) + (bytes->rows plain)) + (replay (cdr xs) + (log-sha256 (bv-append prev sealed)) + (+ expected-gen 1) + (+ count 1)))) + (begin + (unless (and (= count delta-count) + (bv-equal? prev chain-hash)) + (error 'logdb-open + "incremental log delta chain verification failed" + path count delta-count)) + (let-values ([(rows payload) (database-retention-stats db)]) + (let ([log + (make-jlog path key salt head-generation db #f #f + rows payload 0 0 (real-time) + (make-mutex) #f #f (box #f) + #t '() chain-hash delta-count + ckpt-generation 0 #f)]) + (start-jlog-flusher! log) + (make-logdb-handle 'jsqlite log))))))))))))) + ;; Fail closed: a wrong passphrase or a rolled-back/truncated container raises ;; a distinct error instead of returning #f, so auto-mode never silently falls ;; back to an unencrypted legacy log. Only a genuine non-container file (legacy @@ -420,7 +751,9 @@ path) (new-jlog path passphrase))] [(jlog-container-bytes? bytes) - (open-jlog-container path passphrase bytes)] + (if (file-exists? (checkpoint-path path)) + (open-incremental-jlog path passphrase bytes) + (open-jlog-container path passphrase bytes))] [else #f])) (new-jlog path passphrase)))) @@ -505,7 +838,86 @@ (list (cons 'path (jlog-path log)) (cons 'ms (elapsed-ms started))))))) + ;; Write a whole-db checkpoint to path.ckpt and reset the delta chain. This is + ;; the only incremental-mode path that serializes the whole db (O(db-size)); it + ;; runs at creation, periodically, on prune, and for batch/close modes. + (def (write-checkpoint! log generation) + (let* ([plain (sqlite-db->bytevector (jlog-db log))] + [container (encrypt-container (jlog-key log) (jlog-salt log) + generation plain)] + [chain-hash (checkpoint-chain-hash container)]) + (jlog-full-seal-count-set! log (+ (jlog-full-seal-count log) 1)) + (write-file-atomic! (checkpoint-path (jlog-path log)) container) + (write-file-atomic! (delta-path (jlog-path log)) (make-bytevector 0 0)) + (jlog-ckpt-generation-set! log generation) + (jlog-last-seal-hash-set! log chain-hash) + (jlog-delta-frames-set! log 0) + (jlog-pending-data-set! log '()) + (jlog-pruned-since-seal?-set! log #f) + chain-hash)) + + (def (write-head! log generation ckpt-generation delta-count row-count + chain-hash) + (let* ([head-state (pack-head-state ckpt-generation delta-count row-count + chain-hash)] + [sealed (seal-incremental (jlog-key log) *jlog-head-tag* + (jlog-salt log) generation head-state + (make-bytevector 0 0))] + [container (pack-incremental-container (jlog-salt log) generation + sealed)]) + (write-file-atomic! (jlog-path log) container))) + + (def (finish-incremental-persist! log generation) + (let ([path (jlog-path log)]) + (write-generation-marker! path generation) + (jlog-generation-set! log generation) + (jlog-dirty?-set! log #f) + (jlog-pending-rows-set! log 0) + (jlog-pending-bytes-set! log 0) + (jlog-last-persist-ms-set! log (real-time)))) + + ;; Full checkpoint persist for an incremental log (close / batch flusher). + (def (persist-incremental-checkpoint! log) + (let ([generation (+ (jlog-generation log) 1)]) + (write-checkpoint! log generation) + (write-head! log generation (jlog-ckpt-generation log) 0 + (jlog-row-count log) (jlog-last-seal-hash log)) + (finish-incremental-persist! log generation))) + + ;; Immediate-mode persist for an incremental log. Normally appends a single + ;; sealed delta frame for the new rows (O(delta)); consolidates via a full + ;; checkpoint when a prune happened (prunes rewrite arbitrary pages) or the + ;; chain reached the configured length. + (def (persist-incremental-jlog! log) + (let ([generation (+ (jlog-generation log) 1)]) + (if (or (jlog-pruned-since-seal? log) + (>= (jlog-delta-frames log) (log-checkpoint-every))) + (begin + (write-checkpoint! log generation) + (write-head! log generation (jlog-ckpt-generation log) 0 + (jlog-row-count log) (jlog-last-seal-hash log)) + (finish-incremental-persist! log generation)) + (let* ([rows (reverse (jlog-pending-data log))] + [plain (rows->bytes rows)] + [prev-hash (jlog-last-seal-hash log)] + [sealed (seal-incremental (jlog-key log) *jlog-frame-tag* + (jlog-salt log) generation plain + prev-hash)] + [chain-hash (log-sha256 (bv-append prev-hash sealed))]) + (append-delta-frame! (jlog-path log) generation sealed) + (jlog-last-seal-hash-set! log chain-hash) + (jlog-delta-frames-set! log (+ (jlog-delta-frames log) 1)) + (jlog-pending-data-set! log '()) + (write-head! log generation (jlog-ckpt-generation log) + (jlog-delta-frames log) (jlog-row-count log) chain-hash) + (finish-incremental-persist! log generation))))) + (def (persist-jlog! log) + (if (jlog-incremental? log) + (persist-incremental-checkpoint! log) + (persist-jlog-legacy! log))) + + (def (persist-jlog-legacy! log) (let ([total-start (real-time)] [path (jlog-path log)]) (trace-public-event! @@ -674,6 +1086,9 @@ (jlog-payload-bytes-set! log (max 0 (- (jlog-payload-bytes log) (oldest-row-bytes row)))) + ;; A prune rewrites arbitrary pages, so the next immediate persist + ;; must consolidate via a full checkpoint (keeps deltas insert-only). + (jlog-pruned-since-seal?-set! log #t) #t)))) (def (ensure-retention-capacity! log incoming-bytes) @@ -763,11 +1178,13 @@ (if (not (ensure-retention-capacity! log retained-bytes)) #f (begin - (jlog-put-row log (now-seconds) account direction conversation - sender timestamp kind body raw retained-bytes) - (if (persist-on-write?) - (persist-jlog! log) - (begin + (jlog-put-row log (now-seconds) account direction conversation + sender timestamp kind body raw retained-bytes) + (if (persist-on-write?) + (if (jlog-incremental? log) + (persist-incremental-jlog! log) + (persist-jlog-legacy! log)) + (begin (trace-public-event! "logdb-jsqlite-put-deferred" (list (cons 'path (jlog-path log)) @@ -789,6 +1206,11 @@ (empty->null kind) (empty->null body) (or raw "")) + (when (jlog-incremental? log) + (jlog-pending-data-set! log + (cons (list logged-at account direction conversation sender + timestamp kind body raw) + (jlog-pending-data log)))) (jlog-dirty?-set! log #t) (jlog-row-count-set! log (+ (jlog-row-count log) 1)) (jlog-payload-bytes-set! @@ -1011,6 +1433,19 @@ (and (= (bytevector-u8-ref key i) 0) (loop (+ i 1))))))))) + ;; Audit/test hooks: how many full whole-db reseals this handle performed + ;; (immediate mode should keep this flat while delta frames accumulate) and how + ;; many sealed delta frames are pending since the last checkpoint. + (def (logdb-handle-full-seal-count handle) + (and (logdb-handle? handle) + (eq? (logdb-handle-backend handle) 'jsqlite) + (jlog-full-seal-count (logdb-handle-inner handle)))) + + (def (logdb-handle-delta-frame-count handle) + (and (logdb-handle? handle) + (eq? (logdb-handle-backend handle) 'jsqlite) + (jlog-delta-frames (logdb-handle-inner handle)))) + (def (logdb-put handle account direction conversation sender timestamp kind body raw) (and (logdb-handle? handle) new file mode 100644 --- /dev/null +++ b/tests/test-logdb-incremental.ss @@ -0,0 +1,170 @@ +#!chezscheme +;;; Tests for incremental append-only sealing in immediate mode: O(delta) +;;; persists (no whole-db reseal per message), delta-chain replay on open, and +;;; hash-chain tamper/rollback detection. + +(import (except (scheme) + make-hash-table hash-table? + sort sort! + printf fprintf + path-extension path-absolute? + with-input-from-string with-output-to-string + iota 1+ 1- + partition + make-date make-time) + (except (jerboa prelude) meta atom?) + (only (std security taint) safe-delete-file) + (signal logdb)) + +(def (check label pred) + (unless pred + (error 'test-logdb-incremental label))) + +(def (delete-if-exists! p) + (when (file-exists? p) (safe-delete-file p))) + +(def (read-file-bytevector p) + (let ([in (open-file-input-port p)]) + (let ([bv (get-bytevector-all in)]) + (close-port in) + (if (eof-object? bv) (make-bytevector 0 0) bv)))) + +(def (write-file-bytevector! p bv) + (let ([out (open-file-output-port p (file-options no-fail))]) + (put-bytevector out bv) + (close-port out))) + +(def (sub-bytevector bv start end) + (let ([out (make-bytevector (- end start) 0)]) + (bytevector-copy! bv start out 0 (- end start)) + out)) + +(def (clean! p) + (delete-if-exists! p) + (delete-if-exists! (string-append p ".tmp")) + (delete-if-exists! (string-append p ".gen")) + (delete-if-exists! (string-append p ".ckpt")) + (delete-if-exists! (string-append p ".delta"))) + +;; Immediate mode, with the checkpoint interval and retention set high so the +;; test exercises pure delta-frame persists (no periodic checkpoint, no prune). +(putenv "JERBOA_SIGNAL_LOG_PERSIST" "immediate") +(putenv "JERBOA_SIGNAL_LOG_CHECKPOINT_EVERY" "100000") +(putenv "JERBOA_SIGNAL_LOG_MAX_ROWS" "1000000") +(putenv "JERBOA_SIGNAL_LOG_MAX_PAYLOAD_BYTES" "1073741824") +(putenv "JERBOA_SIGNAL_LOG_MAX_ENTRY_BYTES" "67108864") + +(check "logdb backend available" (logdb-available?)) + +;; (1) Immediate persist is O(delta): the whole db is sealed once (the initial +;; checkpoint) and never re-serialized per message; delta frames accumulate and +;; a concurrent reader replays them. +(let ([p "/tmp/jerboa-signal-logdb-incr-test.db"]) + (clean! p) + (let ([h (logdb-open p "incr horse")]) + (check "incremental open" h) + (check "initial checkpoint is exactly one full seal" + (= (logdb-handle-full-seal-count h) 1)) + (do ([i 0 (+ i 1)]) + ((>= i 50)) + (check "incremental put" + (logdb-put h "acct" "in" "direct:incr" "Alice" + (+ 1000 i) "data" + (string-append "msg-" (number->string i)) "{}"))) + (check "no whole-db reseal per message (O(delta))" + (= (logdb-handle-full-seal-count h) 1)) + (check "sealed delta frames accumulated" + (= (logdb-handle-delta-frame-count h) 50)) + (check "in-memory count after incremental puts" + (= (logdb-count h) 50)) + (let ([reader (logdb-open p "incr horse")]) + (check "concurrent reader replays the delta chain" + (= (logdb-count reader) 50)) + (check "replayed most-recent body is correct" + (equal? (list-ref (car (logdb-recent reader 1)) 5) "msg-49")) + (logdb-close reader)) + (logdb-close h)) + ;; Reopen after close still recovers every row. + (let ([h (logdb-open p "incr horse")]) + (check "reopen after close recovers all rows" (= (logdb-count h) 50)) + (logdb-close h)) + (clean! p)) + +;; (2) Tampering with a sealed delta frame fails closed. +(let ([p "/tmp/jerboa-signal-logdb-incr-tamper-test.db"]) + (clean! p) + (let ([h (logdb-open p "tamper horse")]) + (do ([i 0 (+ i 1)]) + ((>= i 5)) + (logdb-put h "a" "in" "c" "s" (+ 2000 i) "data" + (string-append "t-" (number->string i)) "{}")) + (let* ([dp (string-append p ".delta")] + [bv (read-file-bytevector dp)] + [mid (div (bytevector-length bv) 2)]) + (bytevector-u8-set! bv mid + (bitwise-xor (bytevector-u8-ref bv mid) #xff)) + (write-file-bytevector! dp bv)) + (check "tampered delta frame fails closed" + (guard (e [(condition? e) #t]) + (let ([r (logdb-open p "tamper horse")]) + (when r (logdb-close r)) + #f))) + (logdb-close h)) + (clean! p)) + +;; (3) Removing/truncating a trailing append fails closed. +(let ([p "/tmp/jerboa-signal-logdb-incr-remove-test.db"]) + (clean! p) + (let ([h (logdb-open p "remove horse")]) + (do ([i 0 (+ i 1)]) + ((>= i 5)) + (logdb-put h "a" "in" "c" "s" (+ 3000 i) "data" + (string-append "r-" (number->string i)) "{}")) + (let* ([dp (string-append p ".delta")] + [bv (read-file-bytevector dp)] + [n (bytevector-length bv)]) + (write-file-bytevector! dp (sub-bytevector bv 0 (- n 12)))) + (check "removed trailing append fails closed" + (guard (e [(condition? e) #t]) + (let ([r (logdb-open p "remove horse")]) + (when r (logdb-close r)) + #f))) + (logdb-close h)) + (clean! p)) + +;; (4) Rolling back the head container to an older generation fails closed. +(let ([p "/tmp/jerboa-signal-logdb-incr-rollback-test.db"]) + (clean! p) + (let ([h (logdb-open p "rb horse")]) + (logdb-put h "a" "in" "c" "s" 4000 "data" "old" "{}") + (let ([old-head (read-file-bytevector p)]) + (logdb-put h "a" "in" "c" "s" 4001 "data" "new" "{}") + (write-file-bytevector! p old-head) + (check "rolled-back head container fails closed" + (guard (e [(condition? e) #t]) + (let ([r (logdb-open p "rb horse")]) + (when r (logdb-close r)) + #f)))) + (logdb-close h)) + (clean! p)) + +;; (5) A wrong passphrase still fails closed in immediate mode. +(let ([p "/tmp/jerboa-signal-logdb-incr-authfail-test.db"]) + (clean! p) + (let ([h (logdb-open p "right horse")]) + (logdb-put h "a" "in" "c" "s" 5000 "data" "secret" "{}") + (logdb-close h)) + (check "wrong passphrase raises distinct auth error (fail closed)" + (guard (e [(condition? e) + (and (who-condition? e) + (eq? (condition-who e) 'logdb-passphrase-mismatch))]) + (let ([h (logdb-open p "wrong horse")]) + (when h (logdb-close h)) + #f))) + (clean! p)) + +(putenv "JERBOA_SIGNAL_LOG_PERSIST" "") +(putenv "JERBOA_SIGNAL_LOG_CHECKPOINT_EVERY" "") + +(display "logdb incremental regression ok") +(newline)