Skip to content

Bound FTS maintenance across imports and projection rebuilds #173

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
j2h4u wants to merge 9 commits into from
Closed

Bound FTS maintenance across imports and projection rebuilds #173

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
ee9af6f
Bound event search bulk maintenance
j2h4u Jul 14, 2026
7cd9e40
Bound bulk search recovery lifecycle
j2h4u Jul 14, 2026
d033934
Propagate fatal bulk maintenance failures
j2h4u Jul 14, 2026
0da6a91
fix: defer recoverable bounded import checkpoints
j2h4u Jul 14, 2026
f3d1fa9
test(capture): accept deferred checkpoint imports
j2h4u Jul 14, 2026
181b6f9
Share bounded bulk write transactions
j2h4u Jul 15, 2026
d2f2038
Bound and resume search projection rebuilds
j2h4u Jul 15, 2026
169f3cf
Preserve authoritative Codex session identity
j2h4u Jul 15, 2026
eb1fc59
Reject searches during projection rebuilds
j2h4u Jul 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 8 additions & 6 deletions crates/ctx-cli/src/commands/import/native.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,16 +81,18 @@ pub(crate) fn import_one_source_inner(
let bulk_guard = store.begin_event_search_bulk_mode()?;
let import_result =
import_one_source_inner_batched(store, source, progress, full_rescan, preinventory);
let finish_result = store.finish_event_search_bulk_mode(&bulk_guard);
let summary = match (import_result, finish_result) {
let import_result = import_result.and_then(|summary| {
if refresh_search_after_import {
store.refresh_search_index()?;
}
Ok(summary)
});
let finish_result = store.defer_event_search_bulk_mode(&bulk_guard);
match (import_result, finish_result) {
(Ok(summary), Ok(())) => Ok(summary),
(_, Err(error)) => Err(error.into()),
(Err(error), Ok(())) => Err(error),
}?;
if refresh_search_after_import {
store.refresh_search_index()?;
}
Ok(summary)
}

fn import_one_source_inner_batched(
Expand Down
8 changes: 7 additions & 1 deletion crates/ctx-history-capture/src/provider/codex/fast_import.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ pub(crate) fn import_codex_session_paths_fast(
let bulk_guard = store.begin_event_search_bulk_mode()?;
let import_result =
import_codex_session_paths_fast_bounded(paths, store, &options, skipped_by_bounds);
let finish_result = store.finish_event_search_bulk_mode(&bulk_guard);
let finish_result = store.defer_event_search_bulk_mode(&bulk_guard);
match (import_result, finish_result) {
(Ok(summary), Ok(())) => Ok(summary),
(_, Err(err)) => Err(err.into()),
Expand Down Expand Up @@ -222,6 +222,7 @@ pub(crate) fn import_codex_session_path_fast(
.map(|path| path.display().to_string());

let mut header = None;
let mut session_meta_seen = false;
let mut header_persisted = false;
let mut call_contexts: BTreeMap<String, CodexToolCallContext> = BTreeMap::new();
let mut line_number = 0usize;
Expand Down Expand Up @@ -271,6 +272,10 @@ pub(crate) fn import_codex_session_path_fast(
.and_then(Value::as_str)
.unwrap_or("unknown");
if entry_type == "session_meta" {
if session_meta_seen {
continue;
}
session_meta_seen = true;
match codex_session_header(value) {
Ok(parsed) => {
call_contexts.clear();
Expand All @@ -283,6 +288,7 @@ pub(crate) fn import_codex_session_path_fast(
line: line_number,
error: codex_session_file_failure(path, err.to_string()),
});
return Ok(());
}
}
continue;
Expand Down
20 changes: 11 additions & 9 deletions crates/ctx-history-capture/src/provider/codex/session.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ impl ProviderCaptureAdapter for CodexSessionJsonlAdapter {
let mut reader = BufReader::new(file);
let mut result = ProviderNormalizationResult::default();
let mut header = None;
let mut session_meta_seen = false;
let mut call_contexts: BTreeMap<String, CodexToolCallContext> = BTreeMap::new();
let mut has_real_message_content = false;
let mut skipped_oversized_events = 0usize;
Expand Down Expand Up @@ -113,6 +114,10 @@ impl ProviderCaptureAdapter for CodexSessionJsonlAdapter {
.and_then(Value::as_str)
.unwrap_or("unknown");
if entry_type == "session_meta" {
if session_meta_seen {
continue;
}
session_meta_seen = true;
match codex_session_header(value) {
Ok(parsed) => {
let capture = codex_session_capture(
Expand All @@ -132,6 +137,7 @@ impl ProviderCaptureAdapter for CodexSessionJsonlAdapter {
line: line_number,
error: err.to_string(),
});
return Ok(result);
}
}
continue;
Expand Down Expand Up @@ -413,16 +419,12 @@ pub fn import_codex_session_jsonl(
if options.fast_event_inserts {
return import_codex_session_paths_fast(vec![path.to_path_buf()], store, options, 0);
}
let source_path = options
.source_path
.clone()
.unwrap_or_else(|| path.to_path_buf());
let normalization = CodexSessionJsonlAdapter.normalize_path(
path,
&ProviderAdapterContext {
machine_id: options.machine_id,
source_path: Some(source_path),
source_root: None,
source_path: Some(path.to_path_buf()),
source_root: options.source_path.clone(),
imported_at: options.imported_at,
},
)?;
Expand Down Expand Up @@ -460,7 +462,7 @@ pub fn import_codex_session_jsonl_tail(
let bulk_guard = store.begin_event_search_bulk_mode()?;
let import_result =
import_codex_session_jsonl_tail_bounded(path, start_offset, total_bytes, store, &options);
let finish_result = store.finish_event_search_bulk_mode(&bulk_guard);
let finish_result = store.defer_event_search_bulk_mode(&bulk_guard);
match (import_result, finish_result) {
(Ok(summary), Ok(())) => Ok(summary),
(_, Err(err)) => Err(err.into()),
Expand All @@ -480,7 +482,7 @@ fn import_codex_session_jsonl_tail_bounded(
let context = ProviderAdapterContext {
machine_id: options.machine_id.clone(),
source_path: Some(path.to_path_buf()),
source_root: None,
source_root: options.source_path.clone(),
imported_at: options.imported_at,
};
let import_options = NormalizedProviderImportOptions {
Expand Down Expand Up @@ -783,7 +785,7 @@ pub(crate) fn import_codex_session_paths_parallel_normalized(
&options,
skipped_by_bounds,
);
let finish_result = store.finish_event_search_bulk_mode(&bulk_guard);
let finish_result = store.defer_event_search_bulk_mode(&bulk_guard);
match (import_result, finish_result) {
(Ok(summary), Ok(())) => Ok(summary),
(_, Err(err)) => Err(err.into()),
Expand Down
100 changes: 4 additions & 96 deletions crates/ctx-history-capture/src/provider/importer/batches.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,12 @@ use std::{
num::NonZeroUsize,
};

use ctx_history_store::{BoundedBulkWriteTransaction, BULK_WRITE_BATCH_UNITS};
use serde::Serialize;

use super::*;

pub(crate) const IMPORT_TRANSACTION_BATCH_BYTES: usize = 8 * 1024 * 1024;
pub(crate) const IMPORT_TRANSACTION_BATCH_UNITS: usize = 64;
pub(crate) const IMPORT_TRANSACTION_BATCH_UNITS: usize = BULK_WRITE_BATCH_UNITS;

pub(super) fn import_normalized_provider_captures(
store: &mut Store,
Expand Down Expand Up @@ -155,7 +155,7 @@ fn import_provider_capture_lines_with_batch_size(
);
let finish_result = match &bulk_search_guard {
Some(guard) => store
.finish_event_search_bulk_mode(guard)
.defer_event_search_bulk_mode(guard)
.map_err(CaptureError::from),
None => Ok(()),
};
Expand Down Expand Up @@ -314,96 +314,4 @@ impl Write for ByteCounter {
}
}

pub(crate) struct ProviderImportTransaction {
active: bool,
batch_size: Option<NonZeroUsize>,
units: usize,
bytes: usize,
}

impl ProviderImportTransaction {
fn begin(store: &Store, has_work: bool, batch_size: Option<NonZeroUsize>) -> Result<Self> {
if has_work {
store.begin_immediate_batch()?;
}
Ok(Self {
active: has_work,
batch_size,
units: 0,
bytes: 0,
})
}

pub(crate) fn begin_bounded(store: &Store, has_work: bool) -> Result<Self> {
Self::begin(store, has_work, provider_transaction_batch_size())
}

pub(crate) fn prepare_unit(&mut self, store: &Store, unit_bytes: usize) -> Result<()> {
let result = if self.active
&& self.batch_size.is_some()
&& self.units > 0
&& self.bytes.saturating_add(unit_bytes) > IMPORT_TRANSACTION_BATCH_BYTES
{
self.rotate(store)
} else {
Ok(())
};
if result.is_err() {
self.rollback(store);
}
result
}

pub(crate) fn record_unit(&mut self, store: &Store, unit_bytes: usize) -> Result<()> {
if !self.active {
return Ok(());
}
self.units = self.units.saturating_add(1);
self.bytes = self.bytes.saturating_add(unit_bytes);
let below_unit_limit = self
.batch_size
.is_none_or(|batch_size| self.units < batch_size.get());
let below_byte_limit =
self.batch_size.is_none() || self.bytes < IMPORT_TRANSACTION_BATCH_BYTES;
if below_unit_limit && below_byte_limit {
return Ok(());
}
let result = self.rotate(store);
if result.is_err() {
self.rollback(store);
}
result
}

fn rotate(&mut self, store: &Store) -> Result<()> {
store.commit_batch()?;
self.active = false;
store.checkpoint_wal_truncate_required()?;
store.begin_immediate_batch()?;
self.active = true;
self.units = 0;
self.bytes = 0;
Ok(())
}

pub(crate) fn commit(&mut self, store: &Store) -> Result<()> {
let result = if self.active {
store.commit_batch().map_err(CaptureError::from)
} else {
Ok(())
};
if result.is_ok() {
self.active = false;
} else {
self.rollback(store);
}
result
}

pub(crate) fn rollback(&mut self, store: &Store) {
if self.active {
let _ = store.rollback_batch();
self.active = false;
}
}
}
pub(crate) type ProviderImportTransaction = BoundedBulkWriteTransaction;
Loading