Skip to content

Commit dfd29ac

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 974b3ea commit dfd29ac

17 files changed

Lines changed: 394 additions & 286 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
@@ -387,7 +387,10 @@ async fn intent_execution_preserves_recovery_and_completion() {
387387
let cleaned = preparator.cleaned.lock().unwrap();
388388
assert!(cleaned.iter().all(|(_, close_buffers)| *close_buffers == succeeds));
389389
assert_eq!(
390-
cleaned.iter().filter(|(keys, _)| keys.contains(&RESERVED_ALT_KEY)).count(),
390+
cleaned
391+
.iter()
392+
.filter(|(resources, _)| resources.alt_reservation.contains(&RESERVED_ALT_KEY))
393+
.count(),
391394
expected_tasks.len()
392395
);
393396
}

‎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: 5 additions & 8 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,28 +88,25 @@ 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
9594
.optimized_tasks
9695
.iter()
9796
.any(|task| matches!(task, BaseTaskImpl::Undelegate(_))) =>
9897
{
99-
let mut removed = handle_undelegation_error(&self.authority.pubkey(), strategy);
10098
// Report only discarded actions; retained actions still need to execute.
101-
let callbacks = removed.extract_action_callbacks();
99+
let callbacks = handle_undelegation_error(&self.authority.pubkey(), strategy);
102100
if !callbacks.is_empty() {
103101
report.add_callback_report(self.callback_scheduler.schedule(
104102
callbacks,
105103
err.signature(),
106104
Err(ActionError::from(err)),
107105
));
108106
}
109-
removed
110107
}
111108
_ => return Ok(ControlFlow::Break(())),
112109
};
113-
Ok(ControlFlow::Continue(cleanup))
110+
Ok(ControlFlow::Continue(()))
114111
}
115112
}

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

Lines changed: 17 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
use std::{mem, ops::ControlFlow, time::Duration};
1+
use std::{mem, time::Duration};
22

33
use mbv_core::traits::{ActionError, ActionsCallbackScheduler};
44
use program::outbox::PendingTransaction;
@@ -32,7 +32,7 @@ use crate::{
3232
task_info_fetcher::TaskInfoFetcher,
3333
task_strategist::{TransactionPlan, TransactionStrategy},
3434
},
35-
transaction_preparator::TransactionPreparator,
35+
transaction_preparator::{TransactionPreparator, cleanup_resources::CleanupResources},
3636
};
3737

3838
#[cfg(test)]
@@ -106,14 +106,13 @@ where
106106
&& self.has_tasks_after_commit()
107107
&& err.is_recoverable_by_splitting() =>
108108
{
109-
let (current, follow_up, cleanup) = handle_cpi_limit_error(
109+
let (current, follow_up) = handle_cpi_limit_error(
110110
&self.ctx.authority.pubkey(),
111111
mem::take(&mut self.current),
112112
);
113113
self.current = current;
114114
self.follow_up = Some(follow_up);
115115
self.current_attempt = 0;
116-
self.report.dispose(cleanup);
117116
self.report.add_patched_error(err);
118117
continue;
119118
}
@@ -137,7 +136,7 @@ where
137136
result.as_ref().map(|_| ()).map_err(ActionError::from),
138137
);
139138
}
140-
self.report.dispose(mem::take(&mut self.current));
139+
drop(mem::take(&mut self.current));
141140

142141
match result {
143142
Ok(signature) => {
@@ -158,12 +157,7 @@ where
158157
return Ok(output);
159158
}
160159
}
161-
Err(err) => {
162-
if let Some(follow_up) = self.follow_up.take() {
163-
self.report.dispose(follow_up);
164-
}
165-
return Err(err);
166-
}
160+
Err(err) => return Err(err),
167161
}
168162
}
169163
}
@@ -175,10 +169,10 @@ where
175169
if self.current.has_actions_callbacks()
176170
|| self.follow_up.as_ref().is_some_and(TransactionStrategy::has_actions_callbacks)
177171
{
178-
if let Some(time_left) = time_left {
179-
if let Ok(result) = timeout(time_left, self.execute_current()).await {
180-
return result;
181-
}
172+
if let Some(time_left) = time_left
173+
&& let Ok(result) = timeout(time_left, self.execute_current()).await
174+
{
175+
return result;
182176
}
183177
// Only callbacks expire. Keep actions and resolve any pending transaction
184178
// before preparing another one, since the cancelled send may still land.
@@ -235,12 +229,14 @@ where
235229

236230
self.current_attempt += 1;
237231

238-
// Prepare message
232+
// Keep this attempt's resources outside the cancellable preparation future.
233+
let cleanup = self.report.cleanup.push_mut(CleanupResources::default());
239234
let prepared_transaction = prepare_transaction(
240235
&ctx.intent_client,
241236
&ctx.authority,
242237
&ctx.transaction_preparator,
243238
&mut self.current,
239+
cleanup,
244240
)
245241
.await
246242
.map_err(IntentExecutorError::FailedPreparation)?;
@@ -292,12 +288,10 @@ where
292288
self.report.progress.transaction = TransactionProgress::Rejected(*signature);
293289
}
294290
let flow = recovery.recover(&execution_err, &mut self.current, self.report).await?;
295-
let cleanup = match flow {
296-
ControlFlow::Continue(value) => value,
297-
ControlFlow::Break(()) => return Ok(Err(execution_err)),
298-
};
291+
if flow.is_break() {
292+
return Ok(Err(execution_err));
293+
}
299294
ctx.intent_client.invalidate_cached_blockhash().await;
300-
self.report.dispose(cleanup);
301295

302296
if let Some(commit_signature) = confirmed_commit
303297
&& self.current.optimized_tasks.is_empty()
@@ -345,25 +339,23 @@ where
345339
}
346340

347341
fn execute_callbacks(&mut self, signature: Option<Signature>, result: Result<(), ActionError>) {
348-
let cleanup = handle_actions_result(
342+
handle_actions_result(
349343
&self.ctx.authority.pubkey(),
350344
&self.ctx.actions_callback_executor,
351345
self.report,
352346
&mut self.current,
353347
signature,
354348
result.clone(),
355349
);
356-
self.report.dispose(cleanup);
357350
if let (Err(_), Some(follow_up)) = (&result, &mut self.follow_up) {
358-
let cleanup = handle_actions_result(
351+
handle_actions_result(
359352
&self.ctx.authority.pubkey(),
360353
&self.ctx.actions_callback_executor,
361354
self.report,
362355
follow_up,
363356
signature,
364357
result,
365358
);
366-
self.report.dispose(cleanup);
367359
}
368360
}
369361

0 commit comments

Comments
 (0)