Skip to content

Commit cf236e6

Browse files
committed
refactor(committor): track cleanup resources separately
Replace retained transaction strategies with compact cleanup records for each preparation attempt. Records contain buffer identities, acquired ALT reservations, and the uniqueness nonce used by that attempt. Keep them in the execution report before entering cancellable preparation futures. Release executable tasks and their account payloads after use. Recovery helpers now update remaining work without cloning tasks or returning dummy strategies for cleanup. Rename commit_stage_task to buffer_task to reflect its buffer preparation and closure role. Record ALT acquisitions inside TableMania as they occur, including partial failure and cancellation. Separate attempts retain separate reservations even when their keys match, so cleanup cannot under-release repeated reservations or release a key that an attempt never acquired. Close each recorded buffer at most once after successful intent execution, using its recorded nonce. Failed execution still releases ALT reservations without closing buffers that another attempt could reuse. Cleanup no longer rescans executable tasks to recover resource identities. Adapt the existing lifecycle, nonce-recovery, and cleanup tests. Extend the recovery regression to exercise real TableMania reference counts across repeated reservations, partial failure, and cancellation while preserving another intent's reservation. Check buffer closure and nonce association through the cleanup handle. Validation: make fmt; make ci-fmt; cargo check for the committor and TableMania packages with --tests; the recovery_preserves_progress_and_resources test; cargo clippy --workspace --all-targets -- -D warnings; cargo test --workspace (nextest is unavailable locally); git diff --check. Closes #1842.
1 parent 81b3667 commit cf236e6

17 files changed

Lines changed: 396 additions & 299 deletions

File tree

‎committor/src/intent_engine/intent_execution_engine.rs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -493,6 +493,7 @@ mod tests {
493493
test_utils,
494494
transaction_preparator::{
495495
TransactionPreparator,
496+
cleanup_resources::CleanupResources,
496497
delivery_preparator::{BufferExecutionError, DeliveryPreparatorResult},
497498
error::PreparatorResult,
498499
},
@@ -506,14 +507,15 @@ mod tests {
506507
&self,
507508
_authority: &Keypair,
508509
_transaction_strategy: &mut TransactionStrategy,
510+
_cleanup: &mut CleanupResources,
509511
) -> PreparatorResult<VersionedMessage> {
510512
panic!("mock transaction preparator should not prepare a strategy")
511513
}
512514

513-
async fn cleanup_for_strategy(
515+
async fn cleanup_resources(
514516
&self,
515517
_authority: &Keypair,
516-
_transaction_strategy: &TransactionStrategy,
518+
_resources: &CleanupResources,
517519
_close_buffers: bool,
518520
) -> DeliveryPreparatorResult<(), BufferExecutionError> {
519521
Ok(())

‎committor/src/intent_executor/cleanup_handle.rs‎

Lines changed: 17 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,16 @@
1+
use std::collections::HashSet;
2+
13
use futures_util::future::join_all;
24
use solana_keypair::Keypair;
35

4-
use crate::{
5-
tasks::task_strategist::TransactionStrategy,
6-
transaction_preparator::{TransactionPreparator, delivery_preparator::BufferExecutionError},
6+
use crate::transaction_preparator::{
7+
TransactionPreparator, cleanup_resources::CleanupResources,
8+
delivery_preparator::BufferExecutionError,
79
};
810

911
pub struct CleanupHandle<T> {
1012
authority: Keypair,
11-
junk: Vec<TransactionStrategy>,
13+
resources: Vec<CleanupResources>,
1214
/// When false (execution failed), only releases ALT reservations without
1315
/// closing buffer PDAs — avoids a race condition where a concurrent
1416
/// executor may still be using them.
@@ -19,22 +21,28 @@ pub struct CleanupHandle<T> {
1921
impl<T: TransactionPreparator> CleanupHandle<T> {
2022
pub(crate) fn new(
2123
authority: Keypair,
22-
junk: Vec<TransactionStrategy>,
24+
resources: Vec<CleanupResources>,
2325
close_buffers: bool,
2426
transaction_preparator: T,
2527
) -> Self {
2628
Self {
27-
junk,
29+
resources,
2830
close_buffers,
2931
authority,
3032
transaction_preparator,
3133
}
3234
}
3335

34-
pub async fn clean(self) -> Result<(), BufferExecutionError> {
36+
pub async fn clean(mut self) -> Result<(), BufferExecutionError> {
3537
let close_buffers = self.close_buffers;
36-
let cleanup_futs = self.junk.iter().map(|to_cleanup| {
37-
self.transaction_preparator.cleanup_for_strategy(
38+
// Retries can prepare the same buffer again. Close it once, but release
39+
// every ALT reservation, even when separate attempts used identical keys.
40+
let mut buffers = HashSet::new();
41+
for resources in &mut self.resources {
42+
resources.buffers.retain(|task| buffers.insert((task.pubkey, task.commit_id)));
43+
}
44+
let cleanup_futs = self.resources.iter().map(|to_cleanup| {
45+
self.transaction_preparator.cleanup_resources(
3846
&self.authority,
3947
to_cleanup,
4048
close_buffers,

‎committor/src/intent_executor/executor.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -246,7 +246,7 @@ where
246246

247247
let cleanup_handle = CleanupHandle::new(
248248
self.ctx.authority,
249-
report.junk,
249+
report.cleanup,
250250
result.is_ok(),
251251
self.ctx.transaction_preparator,
252252
);

‎committor/src/intent_executor/executor/tests.rs‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -393,7 +393,10 @@ async fn intent_execution_preserves_recovery_and_completion() {
393393
let cleaned = preparator.cleaned.lock().unwrap();
394394
assert!(cleaned.iter().all(|(_, close_buffers)| *close_buffers == succeeds));
395395
assert_eq!(
396-
cleaned.iter().filter(|(keys, _)| keys.contains(&RESERVED_ALT_KEY)).count(),
396+
cleaned
397+
.iter()
398+
.filter(|(resources, _)| resources.alt_reservation.contains(&RESERVED_ALT_KEY))
399+
.count(),
397400
expected_tasks.len()
398401
);
399402
}

‎committor/src/intent_executor/mod.rs‎

Lines changed: 4 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,8 @@ use crate::{
2727
error::{IntentExecutorError, IntentExecutorResult},
2828
intent_execution_client::IntentExecutionClient,
2929
},
30-
tasks::{task_info_fetcher::CacheTaskInfoFetcher, task_strategist::TransactionStrategy},
30+
tasks::task_info_fetcher::CacheTaskInfoFetcher,
31+
transaction_preparator::cleanup_resources::CleanupResources,
3132
};
3233

3334
#[async_trait]
@@ -116,19 +117,15 @@ impl IntentExecutionResult {
116117
#[derive(Default)]
117118
pub struct IntentExecutionReport {
118119
pub progress: ExecutionProgress,
119-
/// Junk that needs to be cleaned up
120-
junk: Vec<TransactionStrategy>,
120+
/// Resource ownership from each preparation attempt, including failed attempts.
121+
cleanup: Vec<CleanupResources>,
121122
/// Errors we patched trying to recover intent
122123
patched_errors: Vec<TransactionStrategyExecutionError>,
123124
/// Report of scheduled callbacks
124125
callbacks_report: Vec<Result<Signature, CallbackScheduleError>>,
125126
}
126127

127128
impl IntentExecutionReport {
128-
pub fn dispose(&mut self, value: TransactionStrategy) {
129-
self.junk.push(value);
130-
}
131-
132129
pub fn add_patched_error(&mut self, value: TransactionStrategyExecutionError) {
133130
self.patched_errors.push(value);
134131
}
@@ -147,8 +144,4 @@ impl IntentExecutionReport {
147144
) {
148145
self.callbacks_report.extend(values);
149146
}
150-
151-
pub fn junk(&self) -> &Vec<TransactionStrategy> {
152-
&self.junk
153-
}
154147
}

‎committor/src/intent_executor/strategy_executor/recovery.rs‎

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -33,14 +33,14 @@ pub(in crate::intent_executor) struct TransactionRecovery<'a, F, A> {
3333
}
3434

3535
impl<F: TaskInfoFetcher, A: ActionsCallbackScheduler> TransactionRecovery<'_, F, A> {
36-
/// Repairs rejected work and returns resources to clean up before completion.
36+
/// Repairs rejected work; preparation records already own its cleanup resources.
3737
/// Follow-ups cannot repair commits, and standalone actions cannot be replayed.
3838
pub(in crate::intent_executor) async fn recover(
3939
&self,
4040
err: &TransactionStrategyExecutionError,
4141
strategy: &mut TransactionStrategy,
4242
report: &mut IntentExecutionReport,
43-
) -> IntentExecutorResult<ControlFlow<(), TransactionStrategy>> {
43+
) -> IntentExecutorResult<ControlFlow<()>> {
4444
let can_repair_commits = report.progress.confirmed_commit.is_none()
4545
&& !self.committed_pubkeys.is_empty()
4646
&& strategy
@@ -51,7 +51,7 @@ impl<F: TaskInfoFetcher, A: ActionsCallbackScheduler> TransactionRecovery<'_, F,
5151
return Ok(ControlFlow::Break(()));
5252
}
5353

54-
let cleanup = match err {
54+
match err {
5555
TransactionStrategyExecutionError::ActionsError(err, signature) => {
5656
handle_actions_result(
5757
&self.authority.pubkey(),
@@ -88,7 +88,6 @@ impl<F: TaskInfoFetcher, A: ActionsCallbackScheduler> TransactionRecovery<'_, F,
8888
.await
8989
.map_err(IntentExecutorError::FailedPreparation)?
9090
.map_err(IntentExecutorError::FailedExecution)?;
91-
TransactionStrategy::default()
9291
}
9392
TransactionStrategyExecutionError::UndelegationError(_, _)
9493
if strategy
@@ -100,6 +99,6 @@ impl<F: TaskInfoFetcher, A: ActionsCallbackScheduler> TransactionRecovery<'_, F,
10099
}
101100
_ => return Ok(ControlFlow::Break(())),
102101
};
103-
Ok(ControlFlow::Continue(cleanup))
102+
Ok(ControlFlow::Continue(()))
104103
}
105104
}

‎committor/src/intent_executor/strategy_executor/transaction.rs‎

Lines changed: 9 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -99,14 +99,13 @@ where
9999
&& self.has_tasks_after_commit()
100100
&& err.is_recoverable_by_splitting() =>
101101
{
102-
let (current, follow_up, cleanup) = handle_cpi_limit_error(
102+
let (current, follow_up) = handle_cpi_limit_error(
103103
&self.ctx.authority.pubkey(),
104104
mem::take(&mut self.current),
105105
);
106106
self.current = current;
107107
self.follow_up = Some(follow_up);
108108
self.current_attempt = 0;
109-
self.report.dispose(cleanup);
110109
self.report.add_patched_error(err);
111110
continue;
112111
}
@@ -130,7 +129,7 @@ where
130129
result.as_ref().map(|_| ()).map_err(ActionError::from),
131130
);
132131
}
133-
self.report.dispose(mem::take(&mut self.current));
132+
drop(mem::take(&mut self.current));
134133

135134
match result {
136135
Ok(signature) => {
@@ -151,12 +150,7 @@ where
151150
return Ok(output);
152151
}
153152
}
154-
Err(err) => {
155-
if let Some(follow_up) = self.follow_up.take() {
156-
self.report.dispose(follow_up);
157-
}
158-
return Err(err);
159-
}
153+
Err(err) => return Err(err),
160154
}
161155
}
162156
}
@@ -168,10 +162,10 @@ where
168162
if self.current.has_actions_callbacks()
169163
|| self.follow_up.as_ref().is_some_and(TransactionStrategy::has_actions_callbacks)
170164
{
171-
if let Some(time_left) = time_left {
172-
if let Ok(result) = timeout(time_left, self.execute_current()).await {
173-
return result;
174-
}
165+
if let Some(time_left) = time_left
166+
&& let Ok(result) = timeout(time_left, self.execute_current()).await
167+
{
168+
return result;
175169
}
176170
// Only callbacks expire. Keep actions and resolve any pending transaction
177171
// before preparing another one, since the cancelled send may still land.
@@ -236,25 +230,23 @@ where
236230
}
237231

238232
fn execute_callbacks(&mut self, signature: Option<Signature>, result: Result<(), ActionError>) {
239-
let cleanup = handle_actions_result(
233+
handle_actions_result(
240234
&self.ctx.authority.pubkey(),
241235
&self.ctx.actions_callback_executor,
242236
self.report,
243237
&mut self.current,
244238
signature,
245239
result.clone(),
246240
);
247-
self.report.dispose(cleanup);
248241
if let (Err(_), Some(follow_up)) = (&result, &mut self.follow_up) {
249-
let cleanup = handle_actions_result(
242+
handle_actions_result(
250243
&self.ctx.authority.pubkey(),
251244
&self.ctx.actions_callback_executor,
252245
self.report,
253246
follow_up,
254247
signature,
255248
result,
256249
);
257-
self.report.dispose(cleanup);
258250
}
259251
}
260252

0 commit comments

Comments
 (0)