Skip to content

Commit 1b5488b

Browse files
Index imported conversations atomically too
`import_accepted_conversation` wrote accepted closures with raw SQL and never reached the search store, so an imported conversation was saved and acknowledged while search could not find it. That is the one thing #294 says must not happen, on a path its task breakdown never named. `finalize_imported_conversation` already materializes every turn inside one write transaction, so it needed no new ordering — only the existing one. Steps 2 to 4 are extracted from `complete` into `index_and_record`, and both accept paths call it: a completion passes one closure, an import passes every closure it materialized. The whole conversation reaches the store as one transaction carrying one revision, since the turns were authored together and no partial import is meaningful. Imports get their own deadline. An import is bulk and one-shot rather than interactive, so the ceiling sized for a single save does not fit it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01AvEcUBSGpcqCNndYiW4rW9
1 parent 2cec6a9 commit 1b5488b

7 files changed

Lines changed: 344 additions & 53 deletions

File tree

crates/relayer-graph-core/src/graph.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ mod search_index;
88
mod writer;
99

1010
pub use completion::{AcceptedGraphClosure, CompletionOutput};
11-
pub use database::{DEFAULT_SEARCH_INDEX_BUDGET, GraphDatabase};
11+
pub use database::{DEFAULT_IMPORT_INDEX_BUDGET, DEFAULT_SEARCH_INDEX_BUDGET, GraphDatabase};
1212
pub use import::{
1313
ImportedAcceptedView, ImportedAction, ImportedConversation, ImportedConversationReceipt,
1414
ImportedConversationStage, ImportedEdge, ImportedInteractionContext, ImportedInvokeOrigin,

crates/relayer-graph-core/src/graph/completion.rs

Lines changed: 56 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -98,34 +98,20 @@ pub(crate) async fn complete(
9898
let closure = read_accepted_closure_on(&mut transaction, scope, scope.root_node_id)
9999
.await?
100100
.ok_or_else(|| GraphError::Internal("accepted closure could not be read".into()))?;
101-
// The next revision has to clear both sides, not just SQLite. A write
102-
// interrupted after the store committed and before SQLite did leaves the
103-
// store ahead; allocating from SQLite alone would hand the next closure a
104-
// number the store has already used for different content.
105-
let recorded = SearchIndexTable::new(&mut transaction)
106-
.revision(target)
107-
.await?;
108-
let revision = match index_revision(database, target, recorded).await {
109-
Ok(revision) => revision,
110-
Err(error) => {
111-
transaction.rollback().await?;
112-
return Err(error);
113-
}
114-
};
115-
116-
// Steps 2 and 3.
117-
let committed = match index_closure(database, target, revision, closure).await {
118-
Ok(committed) => committed,
119-
Err(error) => {
120-
transaction.rollback().await?;
121-
return Err(error);
122-
}
123-
};
124101

125-
// Step 4.
126-
SearchIndexTable::new(&mut transaction)
127-
.record_revision(target, committed)
128-
.await?;
102+
// Steps 2, 3 and 4.
103+
if let Err(error) = index_and_record(
104+
database,
105+
&mut transaction,
106+
target,
107+
vec![closure],
108+
database.expiry(),
109+
)
110+
.await
111+
{
112+
transaction.rollback().await?;
113+
return Err(error);
114+
}
129115
// Step 5.
130116
transaction.commit().await?;
131117
// Step 6.
@@ -134,40 +120,66 @@ pub(crate) async fn complete(
134120
.ok_or_else(|| GraphError::Internal("accepted completion could not be read".into()))
135121
}
136122

137-
/// Write and commit one closure to the search store, under a deadline.
123+
/// Write closures to the search store, commit them, and record the revision in
124+
/// the caller's still-open SQLite transaction — steps 2 to 4 of the ordering.
138125
///
139-
/// The deadline is required rather than defensive: the SQLite write lock is held
140-
/// across this call, so an unbounded search write would stall every other writer
141-
/// in the database. On timeout the search transaction is rolled back and the
142-
/// write fails, leaving nothing committed anywhere.
143-
async fn index_revision(
126+
/// Both accept paths share this. A completion passes one closure; an import
127+
/// passes every closure it materialized, so a whole conversation reaches the
128+
/// store as one transaction carrying one revision.
129+
///
130+
/// The caller owns the SQLite transaction and must roll it back on error.
131+
/// Nothing here commits it.
132+
pub(crate) async fn index_and_record(
144133
database: &GraphDatabase,
134+
transaction: &mut GraphConnection,
145135
target: SearchTarget,
146-
recorded: Option<SearchIndexRevision>,
147-
) -> Result<SearchIndexRevision, GraphError> {
148-
let stored = deadline(database.expiry(), database.search_index.revision(target)).await?;
149-
Ok(recorded
136+
closures: Vec<AcceptedGraphClosure>,
137+
expiry: tokio::time::Instant,
138+
) -> Result<(), GraphError> {
139+
// The next revision has to clear both sides, not just SQLite. A write
140+
// interrupted after the store committed and before SQLite did leaves the
141+
// store ahead; allocating from SQLite alone would hand these closures a
142+
// number the store has already used for different content.
143+
let recorded = SearchIndexTable::new(&mut *transaction)
144+
.revision(target)
145+
.await?;
146+
let stored = deadline(expiry, database.search_index.revision(target)).await?;
147+
let revision = recorded
150148
.max(stored)
151-
.map_or(SearchIndexRevision::FIRST, SearchIndexRevision::next))
149+
.map_or(SearchIndexRevision::FIRST, SearchIndexRevision::next);
150+
151+
let committed = index_closures(database, target, revision, closures, expiry).await?;
152+
SearchIndexTable::new(&mut *transaction)
153+
.record_revision(target, committed)
154+
.await?;
155+
Ok(())
152156
}
153157

154-
async fn index_closure(
158+
/// Write and commit closures to the search store, under a deadline.
159+
///
160+
/// The deadline is required rather than defensive: the SQLite write lock is held
161+
/// across this call, so an unbounded search write would stall every other writer
162+
/// in the database. On timeout the search transaction is rolled back and the
163+
/// write fails, leaving nothing committed anywhere.
164+
async fn index_closures(
155165
database: &GraphDatabase,
156166
target: SearchTarget,
157167
revision: SearchIndexRevision,
158-
closure: AcceptedGraphClosure,
168+
closures: Vec<AcceptedGraphClosure>,
169+
expiry: tokio::time::Instant,
159170
) -> Result<SearchIndexRevision, GraphError> {
160171
// One deadline spans the whole sequence rather than each step, because what
161172
// is being bounded is how long the global SQLite write lock is held. A
162173
// per-step budget would let a slow store hold it for a multiple of it.
163-
let expiry = database.expiry();
164174
let mut write = deadline(expiry, database.search_index.begin(target, revision)).await?;
165175
// Past this point a failure has to release the search transaction, or the
166176
// store keeps its write lock and every later write fails behind it. An
167177
// abandoned write releases its own transaction when it is dropped.
168-
if let Err(error) = deadline(expiry, write.apply(closure)).await {
169-
let _ = deadline(expiry, write.rollback()).await;
170-
return Err(error);
178+
for closure in closures {
179+
if let Err(error) = deadline(expiry, write.apply(closure)).await {
180+
let _ = deadline(expiry, write.rollback()).await;
181+
return Err(error);
182+
}
171183
}
172184
// A commit that outlives the deadline is not rolled back, because by then it
173185
// may already have committed. The caller fails the write and rolls SQLite

crates/relayer-graph-core/src/graph/database.rs

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,11 +27,19 @@ use crate::{
2727
/// the SQLite busy timeout, and #303 measures the latency that actually matters.
2828
pub const DEFAULT_SEARCH_INDEX_BUDGET: Duration = Duration::from_secs(5);
2929

30+
/// How long indexing an imported conversation may take before the import fails.
31+
///
32+
/// An import is bulk and one-shot rather than interactive: it materializes every
33+
/// turn of a conversation and indexes them as one transaction, so it needs a
34+
/// ceiling of its own rather than the one sized for a single save.
35+
pub const DEFAULT_IMPORT_INDEX_BUDGET: Duration = Duration::from_secs(60);
36+
3037
#[derive(Clone)]
3138
pub struct GraphDatabase {
3239
pub(crate) storage: SqliteGraphStore,
3340
pub(crate) search_index: Arc<dyn SearchIndex>,
3441
pub(crate) search_index_budget: Duration,
42+
pub(crate) import_index_budget: Duration,
3543
/// One lock per logical target, so concurrent submissions to one target
3644
/// index in the order they commit while unrelated targets stay independent.
3745
write_order: Arc<Mutex<HashMap<SearchTarget, Arc<AsyncMutex<()>>>>>,
@@ -54,6 +62,7 @@ impl GraphDatabase {
5462
storage: SqliteGraphStore::open(path).await?,
5563
search_index,
5664
search_index_budget: DEFAULT_SEARCH_INDEX_BUDGET,
65+
import_index_budget: DEFAULT_IMPORT_INDEX_BUDGET,
5766
write_order: Arc::default(),
5867
})
5968
}
@@ -69,6 +78,7 @@ impl GraphDatabase {
6978
storage: SqliteGraphStore::in_memory().await?,
7079
search_index,
7180
search_index_budget: DEFAULT_SEARCH_INDEX_BUDGET,
81+
import_index_budget: DEFAULT_IMPORT_INDEX_BUDGET,
7282
write_order: Arc::default(),
7383
})
7484
}
@@ -92,6 +102,17 @@ impl GraphDatabase {
92102
tokio::time::Instant::now() + self.search_index_budget
93103
}
94104

105+
/// The deadline for indexing a whole imported conversation.
106+
pub(crate) fn import_expiry(&self) -> tokio::time::Instant {
107+
tokio::time::Instant::now() + self.import_index_budget
108+
}
109+
110+
/// Bound how long indexing an imported conversation may take.
111+
pub fn with_import_index_budget(mut self, budget: Duration) -> Self {
112+
self.import_index_budget = budget;
113+
self
114+
}
115+
95116
/// Take this target's place in line. Submissions to one target are ordered
96117
/// against each other and against nothing else.
97118
pub(crate) async fn order_writes_to(&self, target: SearchTarget) -> OwnedMutexGuard<()> {

crates/relayer-graph-core/src/graph/import.rs

Lines changed: 50 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,10 @@ use std::collections::{HashMap, HashSet};
22

33
use serde::{Deserialize, Serialize};
44

5-
use crate::{GraphError, PERSONAL_PRESENTATION_PROFILE_THREAD_ID, ProjectId, ThreadId};
5+
use crate::{
6+
GraphError, PERSONAL_PRESENTATION_PROFILE_THREAD_ID, ProjectId, ThreadId,
7+
graph::{InteractionScope, completion},
8+
};
69

710
#[derive(Debug, Clone, Deserialize, Serialize)]
811
#[serde(rename_all = "camelCase")]
@@ -215,6 +218,21 @@ impl crate::GraphDatabase {
215218
&self,
216219
import_id: &str,
217220
) -> Result<ImportedConversationReceipt, GraphError> {
221+
// Taken before the write transaction, like every other accept path, so
222+
// submissions to this target index in the order they commit.
223+
let target = {
224+
let mut connection = self.storage.acquire().await?;
225+
let (project_id, thread_id): (Option<i64>, i64) =
226+
sqlx::query_as("SELECT project_id,thread_id FROM graph_imports WHERE import_id=?1")
227+
.bind(import_id)
228+
.fetch_one(&mut *connection)
229+
.await?;
230+
let thread_id = ThreadId::new(thread_id).ok_or_else(|| {
231+
GraphError::Internal("imported conversation has an invalid thread".into())
232+
})?;
233+
crate::SearchTarget::new(project_id.and_then(ProjectId::new), thread_id)
234+
};
235+
let _order = self.order_writes_to(target).await;
218236
let mut tx = self.storage.begin_write().await?;
219237
let metadata = load_metadata(&mut tx, import_id).await?;
220238
let turn_count: i64 =
@@ -620,6 +638,37 @@ impl crate::GraphDatabase {
620638
));
621639
}
622640
}
641+
// An import is an accept path like any other, so its closures reach the
642+
// search store before SQLite commits. The whole conversation goes in as
643+
// one search transaction carrying one revision: the turns were authored
644+
// together and there is no point at which a partial import is meaningful.
645+
let mut closures = Vec::new();
646+
for receipt in &receipts {
647+
let Some(node_id) = receipt.graph_node_id else {
648+
continue;
649+
};
650+
let node_id = crate::NodeId::new(node_id)
651+
.ok_or_else(|| GraphError::Internal("invalid imported root node ID".into()))?;
652+
let scope = InteractionScope {
653+
project_id: metadata.project_id,
654+
thread_id: metadata.thread_id,
655+
root_node_id: node_id,
656+
read_only: false,
657+
};
658+
if let Some(closure) =
659+
completion::read_accepted_closure_on(&mut tx, &scope, node_id).await?
660+
{
661+
closures.push(closure);
662+
}
663+
}
664+
if !closures.is_empty()
665+
&& let Err(error) =
666+
completion::index_and_record(self, &mut tx, target, closures, self.import_expiry())
667+
.await
668+
{
669+
tx.rollback().await?;
670+
return Err(error);
671+
}
623672
tx.commit().await?;
624673
for receipt in &mut receipts {
625674
if let Some(node_id) = receipt.graph_node_id {

crates/relayer-graph-core/src/lib.rs

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5,12 +5,12 @@ mod storage;
55
pub use error::{GraphError, ValidationIssue};
66
pub use graph::{
77
AcceptedGraphClosure, ActionDraft, ActionId, ActionKind, ActionVariant, CompletionOutput,
8-
DEFAULT_SEARCH_INDEX_BUDGET, EdgeDraft, EdgeId, GraphAction, GraphDatabase, GraphEdge,
9-
GraphLayer, GraphNode, GraphWriter, ImportedAcceptedView, ImportedAction, ImportedConversation,
10-
ImportedConversationReceipt, ImportedConversationStage, ImportedEdge,
11-
ImportedInteractionContext, ImportedInvokeOrigin, ImportedLayer, ImportedLayerLayout,
12-
ImportedNode, ImportedNodePlacement, ImportedResolvedLayer, ImportedTurn, ImportedTurnReceipt,
13-
InteractionContext, InteractionContextAction, InteractionContextDraft,
8+
DEFAULT_IMPORT_INDEX_BUDGET, DEFAULT_SEARCH_INDEX_BUDGET, EdgeDraft, EdgeId, GraphAction,
9+
GraphDatabase, GraphEdge, GraphLayer, GraphNode, GraphWriter, ImportedAcceptedView,
10+
ImportedAction, ImportedConversation, ImportedConversationReceipt, ImportedConversationStage,
11+
ImportedEdge, ImportedInteractionContext, ImportedInvokeOrigin, ImportedLayer,
12+
ImportedLayerLayout, ImportedNode, ImportedNodePlacement, ImportedResolvedLayer, ImportedTurn,
13+
ImportedTurnReceipt, InteractionContext, InteractionContextAction, InteractionContextDraft,
1414
InteractionContextTarget, InteractionInput, InteractionInputNode, InteractionInvocation,
1515
LayerDraft, LayerId, LayerLayout, NavigateRelation, NoSearchIndex, NodeDraft, NodeId,
1616
NodePlacement, PERSONAL_PRESENTATION_PROFILE_THREAD_ID, PersonalPresentationAttachment,

0 commit comments

Comments
 (0)