agent: pass SharedUpdateBroadcastCache to /updates handler

It's required by the handler but not getting passed to Axum's middleware
stack.

crates/corro-agent/src/agent/run_root.rs

@@ -49,6 +49,7 @@ async fn run(agent: Agent, opts: AgentOptions, pconf: PerfConfig) -> eyre::Resul
         rx_foca,
         subs_manager,
         subs_bcast_cache,
+        updates_bcast_cache,
         rtt_rx,
     } = opts;
 
@@ -96,6 +97,7 @@ async fn run(agent: Agent, opts: AgentOptions, pconf: PerfConfig) -> eyre::Resul
         &agent,
         &tripwire,
         subs_bcast_cache,
+        updates_bcast_cache,
         &subs_manager,
         api_listeners,
     )

crates/corro-agent/src/agent/setup.rs

@@ -25,10 +25,14 @@ use tripwire::Tripwire;
 use crate::{
     api::{
         peer::gossip_server_endpoint,
-        public::pubsub::{process_sub_channel, MatcherBroadcastCache, SharedMatcherBroadcastCache},
+        public::{
+            pubsub::{process_sub_channel, MatcherBroadcastCache, SharedMatcherBroadcastCache},
+            update::SharedUpdateBroadcastCache,
+        },
     },
     transport::Transport,
 };
+use corro_types::updates::UpdatesManager;
 use corro_types::{
     actor::ActorId,
     agent::{migrate, Agent, AgentConfig, Booked, BookedVersions, LockRegistry, SplitPool},
@@ -41,7 +45,6 @@ use corro_types::{
     schema::{init_schema, Schema},
     sqlite::CrConn,
 };
-use corro_types::updates::UpdatesManager;
 
 /// Runtime state for the Corrosion agent
 pub struct AgentOptions {
@@ -58,6 +61,7 @@ pub struct AgentOptions {
     pub rtt_rx: TokioReceiver<(SocketAddr, Duration)>,
     pub subs_manager: SubsManager,
     pub subs_bcast_cache: SharedMatcherBroadcastCache,
+    pub updates_bcast_cache: SharedUpdateBroadcastCache,
     pub tripwire: Tripwire,
 }
 
@@ -118,6 +122,8 @@ pub async fn setup(conf: Config, tripwire: Tripwire) -> eyre::Result<(Agent, Age
     )
     .await?;
 
+    let updates_bcast_cache = SharedUpdateBroadcastCache::default();
+
     let cluster_id = {
         let conn = pool.read().await?;
         conn.query_row(
@@ -189,6 +195,7 @@ pub async fn setup(conf: Config, tripwire: Tripwire) -> eyre::Result<(Agent, Age
         rtt_rx,
         subs_manager: subs_manager.clone(),
         subs_bcast_cache,
+        updates_bcast_cache,
         tripwire: tripwire.clone(),
     };
 

crates/corro-agent/src/agent/util.rs

@@ -10,6 +10,7 @@ use crate::{
     api::public::{
         api_v1_db_schema, api_v1_queries, api_v1_table_stats, api_v1_transactions,
         pubsub::{api_v1_sub_by_id, api_v1_subs},
+        update::SharedUpdateBroadcastCache,
     },
     transport::Transport,
 };
@@ -170,6 +171,7 @@ pub async fn setup_http_api_handler(
     agent: &Agent,
     tripwire: &Tripwire,
     subs_bcast_cache: BcastCache,
+    updates_bcast_cache: SharedUpdateBroadcastCache,
     subs_manager: &SubsManager,
     api_listeners: Vec<TcpListener>,
 ) -> eyre::Result<()> {
@@ -280,6 +282,7 @@ pub async fn setup_http_api_handler(
                 .layer(Extension(Arc::new(AtomicI64::new(0))))
                 .layer(Extension(agent.clone()))
                 .layer(Extension(subs_bcast_cache))
+                .layer(Extension(updates_bcast_cache))
                 .layer(Extension(subs_manager.clone()))
                 .layer(Extension(tripwire.clone())),
         )