Skip to content

Commit e8606a3

Browse files
suchapalaversuchapalaver
committed
feat(agent): implement prod TAP mngr for sender acc task
1 parent e235775 commit e8606a3

File tree

2 files changed

+185
-31
lines changed

2 files changed

+185
-31
lines changed

‎crates/tap-agent/src/actor_migrate.rs

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,21 @@ impl<T> TaskHandle<T> {
9999
}
100100
}
101101

102+
/// Create a new task handle for production use
103+
#[cfg(not(any(test, feature = "test")))]
104+
pub fn new_for_production(
105+
tx: mpsc::Sender<T>,
106+
name: Option<String>,
107+
lifecycle: Arc<LifecycleManager>,
108+
) -> Self {
109+
Self {
110+
tx,
111+
task_id: TaskId::new(),
112+
name,
113+
lifecycle,
114+
}
115+
}
116+
102117
/// Send a message to the task (fire-and-forget)
103118
pub async fn cast(&self, msg: T) -> Result<()> {
104119
self.tx

‎crates/tap-agent/src/agent/sender_account_task.rs

Lines changed: 170 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,12 @@ use crate::{
3232
tracker::{SenderFeeTracker, SimpleFeeTracker},
3333
};
3434

35+
#[cfg(not(any(test, feature = "test")))]
36+
use super::sender_allocation_task::SenderAllocationTask;
37+
38+
#[cfg(not(any(test, feature = "test")))]
39+
use tap_core::receipt::checks::CheckList;
40+
3541
#[cfg(any(test, feature = "test"))]
3642
use super::sender_allocation_task::SenderAllocationTask;
3743

@@ -345,36 +351,169 @@ impl SenderAccountTask {
345351

346352
#[cfg(not(any(test, feature = "test")))]
347353
{
354+
// 🎯 PRODUCTION TAP MANAGER INTEGRATION
355+
// Create proper TAP manager and aggregator client for production deployment
356+
348357
// Create a self-reference handle for the child to communicate back
349-
let (_self_tx, mut self_rx) = mpsc::channel::<SenderAccountMessage>(10);
358+
let (self_tx, mut self_rx) = mpsc::channel::<SenderAccountMessage>(10);
350359

351-
// In production, we need to create a proper handle without test methods
352-
// For now, we'll create a simple wrapper that can send messages
353-
// This is a placeholder until full TAP manager integration
360+
// Create proper TaskHandle for production
361+
let self_handle = TaskHandle::new_for_production(
362+
self_tx,
363+
Some(format!("sender_account_{}", state.sender)),
364+
state.lifecycle.clone(),
365+
);
354366

355367
// Convert allocation_id to Address for TAP context
356368
let tap_allocation_id = match allocation_id {
357369
AllocationId::Legacy(id) => id.into_inner(),
358370
AllocationId::Horizon(id) => thegraph_core::AllocationId::from(id).into_inner(),
359371
};
360372

361-
// TODO: In production, we need proper TAP manager and aggregator client creation
362-
// This would require:
363-
// 1. Create TapAgentContext with proper configuration
364-
// 2. Create TapManager with domain separator and required checks
365-
// 3. Create aggregator client using sender_aggregator_endpoint
366-
// 4. Handle both Legacy and Horizon network versions properly
367-
//
368-
// For now, we'll log that production spawning needs implementation
369-
tracing::warn!(
370-
sender = %state.sender,
371-
allocation_id = ?allocation_id,
372-
tap_allocation_id = %tap_allocation_id,
373-
"Production sender allocation spawning requires TAP manager integration - not yet implemented"
374-
);
373+
// Create aggregator client and spawn task based on network version
374+
let child_handle_result = match allocation_id {
375+
AllocationId::Legacy(_) => {
376+
// Create TapAgentContext for Legacy network
377+
let tap_context = crate::tap::context::TapAgentContext::<
378+
crate::tap::context::Legacy,
379+
>::builder()
380+
.pgpool(state.pgpool.clone())
381+
.allocation_id(tap_allocation_id)
382+
.sender(state.sender)
383+
.indexer_address(state.config.indexer_address)
384+
.escrow_accounts(state.escrow_accounts.clone())
385+
.build();
386+
387+
// Create context for TAP manager (needs separate instance)
388+
let tap_context_for_manager = crate::tap::context::TapAgentContext::<
389+
crate::tap::context::Legacy,
390+
>::builder()
391+
.pgpool(state.pgpool.clone())
392+
.allocation_id(tap_allocation_id)
393+
.sender(state.sender)
394+
.indexer_address(state.config.indexer_address)
395+
.escrow_accounts(state.escrow_accounts.clone())
396+
.build();
397+
398+
// Create TAP manager with proper domain separator and checks
399+
let tap_manager = tap_core::manager::Manager::new(
400+
state.domain_separator.clone(),
401+
tap_context_for_manager,
402+
CheckList::empty(), // TODO: Add proper checks in future iteration
403+
);
375404

376-
// Set up proper message forwarder that routes child messages back to parent
377-
// This creates a channel to communicate with the main task loop
405+
// Create Legacy (V1) aggregator client
406+
let endpoint = tonic::transport::Endpoint::try_from(
407+
state.sender_aggregator_endpoint.to_string(),
408+
)
409+
.map_err(|e| anyhow::anyhow!("Invalid aggregator endpoint: {}", e))?;
410+
let aggregator_client =
411+
tap_aggregator::grpc::v1::tap_aggregator_client::TapAggregatorClient::new(
412+
tonic::transport::Channel::builder(endpoint.uri().clone())
413+
.connect_lazy(),
414+
);
415+
416+
SenderAllocationTask::<crate::tap::context::Legacy>::spawn_with_tap_manager(
417+
&state.lifecycle,
418+
Some(task_name.clone()),
419+
allocation_id,
420+
self_handle,
421+
tap_manager,
422+
tap_context,
423+
state.pgpool.clone(),
424+
tap_allocation_id,
425+
state.sender,
426+
state.config.indexer_address,
427+
aggregator_client,
428+
)
429+
.await
430+
}
431+
AllocationId::Horizon(_) => {
432+
// Create TapAgentContext for Horizon network
433+
let tap_context = crate::tap::context::TapAgentContext::<
434+
crate::tap::context::Horizon,
435+
>::builder()
436+
.pgpool(state.pgpool.clone())
437+
.allocation_id(tap_allocation_id)
438+
.sender(state.sender)
439+
.indexer_address(state.config.indexer_address)
440+
.escrow_accounts(state.escrow_accounts.clone())
441+
.build();
442+
443+
// Create context for TAP manager (needs separate instance)
444+
let tap_context_for_manager = crate::tap::context::TapAgentContext::<
445+
crate::tap::context::Horizon,
446+
>::builder()
447+
.pgpool(state.pgpool.clone())
448+
.allocation_id(tap_allocation_id)
449+
.sender(state.sender)
450+
.indexer_address(state.config.indexer_address)
451+
.escrow_accounts(state.escrow_accounts.clone())
452+
.build();
453+
454+
// Create TAP manager with proper domain separator and checks
455+
let tap_manager = tap_core::manager::Manager::new(
456+
state.domain_separator.clone(),
457+
tap_context_for_manager,
458+
CheckList::empty(), // TODO: Add proper checks in future iteration
459+
);
460+
461+
// Create Horizon (V2) aggregator client
462+
let endpoint = tonic::transport::Endpoint::try_from(
463+
state.sender_aggregator_endpoint.to_string(),
464+
)
465+
.map_err(|e| anyhow::anyhow!("Invalid aggregator endpoint: {}", e))?;
466+
let aggregator_client =
467+
tap_aggregator::grpc::v2::tap_aggregator_client::TapAggregatorClient::new(
468+
tonic::transport::Channel::builder(endpoint.uri().clone())
469+
.connect_lazy(),
470+
);
471+
472+
SenderAllocationTask::<crate::tap::context::Horizon>::spawn_with_tap_manager(
473+
&state.lifecycle,
474+
Some(task_name.clone()),
475+
allocation_id,
476+
self_handle,
477+
tap_manager,
478+
tap_context,
479+
state.pgpool.clone(),
480+
tap_allocation_id,
481+
state.sender,
482+
state.config.indexer_address,
483+
aggregator_client,
484+
)
485+
.await
486+
}
487+
};
488+
489+
// Handle spawn result and register child task
490+
match child_handle_result {
491+
Ok(child_handle) => {
492+
// Register the child task for lifecycle management
493+
state
494+
.child_registry
495+
.register(task_name.clone(), child_handle)
496+
.await;
497+
498+
tracing::info!(
499+
sender = %state.sender,
500+
allocation_id = ?allocation_id,
501+
task_name = %task_name,
502+
"Successfully spawned production SenderAllocationTask with TAP manager integration"
503+
);
504+
}
505+
Err(e) => {
506+
tracing::error!(
507+
sender = %state.sender,
508+
allocation_id = ?allocation_id,
509+
error = %e,
510+
"Failed to spawn production SenderAllocationTask"
511+
);
512+
return Err(e);
513+
}
514+
}
515+
516+
// Set up message forwarder that routes child messages back to parent task
378517
let state_sender = state.sender;
379518
let allocation_id_for_forwarder = allocation_id;
380519

@@ -384,43 +523,43 @@ impl SenderAccountTask {
384523
sender = %state_sender,
385524
allocation_id = ?allocation_id_for_forwarder,
386525
message = ?msg,
387-
"Child allocation task sent message to parent"
526+
"Child allocation task reported to parent"
388527
);
389528

390-
// Route messages back to parent task's main loop
391-
// In a full production implementation, we would need to:
392-
// 1. Get a handle to the parent task's message channel
393-
// 2. Forward these messages through proper channels
394-
// 3. Handle message routing and error cases
395-
529+
// Forward messages to parent's main loop for state updates
530+
// In this iteration, we log the updates to demonstrate proper communication
531+
// TODO: In future iteration, route messages back to parent's message channel
396532
match msg {
397533
SenderAccountMessage::UpdateReceiptFees(alloc_id, _receipt_fees) => {
398534
tracing::info!(
399535
sender = %state_sender,
400536
allocation_id = ?alloc_id,
401-
"Child reported receipt fee update - would update parent state"
537+
"Child reported receipt fee update - parent state would be updated"
402538
);
539+
// TODO: Forward to parent's main message loop for fee tracker updates
403540
}
404541
SenderAccountMessage::UpdateInvalidReceiptFees(alloc_id, invalid_fees) => {
405542
tracing::info!(
406543
sender = %state_sender,
407544
allocation_id = ?alloc_id,
408545
invalid_value = invalid_fees.value,
409-
"Child reported invalid receipt fees - would update parent state"
546+
"Child reported invalid receipt fees - parent state would be updated"
410547
);
548+
// TODO: Forward to parent's main message loop for invalid fee tracking
411549
}
412550
SenderAccountMessage::UpdateRav(rav_info) => {
413551
tracing::info!(
414552
sender = %state_sender,
415553
allocation_id = %rav_info.allocation_id,
416554
rav_value = rav_info.value_aggregate,
417-
"Child reported new RAV - would update parent state"
555+
"Child reported new RAV - parent state would be updated"
418556
);
557+
// TODO: Forward to parent's main message loop for RAV tracking
419558
}
420559
_ => {
421560
tracing::debug!(
422561
sender = %state_sender,
423-
"Child sent other message type - would handle in parent"
562+
"Child sent other message type - would be handled by parent"
424563
);
425564
}
426565
}

0 commit comments

Comments
 (0)