···

       
80
       
 
       
    cursor: Option<Cursor>,

     

       
81
       
 
       
    sketch_secret: SketchSecretPrefix,

     

       
82
       
 
       
) -> anyhow::Result<()> {

     

       
83
       
-
       
    let mut tasks: JoinSet<anyhow::Result<()>> = JoinSet::new();

     

       
0
       
 
       

     

       
84
       
 
       


     

       
85
       
 
       
    println!("starting server with storage...");

     

       
86
       
 
       
    let serving = server::serve(read_store.clone());

     

       
87
       
-
       
    tasks.spawn(async move {

     

       
88
       
 
       
        serving.await.map_err(|e| {

     

       
89
       
 
       
            log::warn!("server ended: {e}");

     

       
90
       
 
       
            anyhow::anyhow!(e)

     
···

       
93
       
 
       


     

       
94
       
 
       
    if args.pause_writer {

     

       
95
       
 
       
        log::info!("not starting jetstream or the write loop.");

     

       
96
       
-
       
        for t in tasks.join_all().await {

     

       
97
       
 
       
            if let Err(e) = t {

     

       
98
       
 
       
                return Err(anyhow::anyhow!(e));

     

       
99
       
 
       
            }

     
···

       
115
       
 
       
    let rolling = write_store

     

       
116
       
 
       
        .background_tasks(args.reroll)?

     

       
117
       
 
       
        .run(args.backfill);

     

       
118
       
-
       
    tasks.spawn(async move {

     

       
119
       
 
       
        rolling

     

       
120
       
 
       
            .await

     

       
121
       
 
       
            .inspect_err(|e| log::warn!("rollup ended: {e}"))?;

     

       
122
       
 
       
        Ok(())

     

       
123
       
 
       
    });

     

       
124
       
 
       


     

       
125
       
-
       
    tasks.spawn(async move {

     

       
126
       
 
       
        write_store

     

       
127
       
 
       
            .receive_batches(batches)

     

       
128
       
 
       
            .await

     
···

       
130
       
 
       
        Ok(())

     

       
131
       
 
       
    });

     

       
132
       
 
       


     

       
133
       
-
       
    tasks.spawn(async move {

     

       
134
       
 
       
        do_update_stuff(read_store).await;

     

       
135
       
 
       
        log::warn!("status task ended");

     

       
136
       
 
       
        Ok(())

     
···

       
138
       
 
       


     

       
139
       
 
       
    install_metrics_server()?;

     

       
140
       
 
       


     

       
141
       
-
       
    for (i, t) in tasks.join_all().await.iter().enumerate() {

     

       
142
       
 
       
        log::warn!("task {i} done: {t:?}");

     

       
143
       
 
       
    }

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
144
       
 
       


     

       
145
       
 
       
    println!("bye!");

     

       
146
       
 
       


     

···

       
80
       
 
       
    cursor: Option<Cursor>,

     

       
81
       
 
       
    sketch_secret: SketchSecretPrefix,

     

       
82
       
 
       
) -> anyhow::Result<()> {

     

       
83
       
+
       
    let mut whatever_tasks: JoinSet<anyhow::Result<()>> = JoinSet::new();

     

       
84
       
+
       
    let mut consumer_tasks: JoinSet<anyhow::Result<()>> = JoinSet::new();

     

       
85
       
 
       


     

       
86
       
 
       
    println!("starting server with storage...");

     

       
87
       
 
       
    let serving = server::serve(read_store.clone());

     

       
88
       
+
       
    whatever_tasks.spawn(async move {

     

       
89
       
 
       
        serving.await.map_err(|e| {

     

       
90
       
 
       
            log::warn!("server ended: {e}");

     

       
91
       
 
       
            anyhow::anyhow!(e)

     
···

       
94
       
 
       


     

       
95
       
 
       
    if args.pause_writer {

     

       
96
       
 
       
        log::info!("not starting jetstream or the write loop.");

     

       
97
       
+
       
        for t in whatever_tasks.join_all().await {

     

       
98
       
 
       
            if let Err(e) = t {

     

       
99
       
 
       
                return Err(anyhow::anyhow!(e));

     

       
100
       
 
       
            }

     
···

       
116
       
 
       
    let rolling = write_store

     

       
117
       
 
       
        .background_tasks(args.reroll)?

     

       
118
       
 
       
        .run(args.backfill);

     

       
119
       
+
       
    consumer_tasks.spawn(async move {

     

       
120
       
 
       
        rolling

     

       
121
       
 
       
            .await

     

       
122
       
 
       
            .inspect_err(|e| log::warn!("rollup ended: {e}"))?;

     

       
123
       
 
       
        Ok(())

     

       
124
       
 
       
    });

     

       
125
       
 
       


     

       
126
       
+
       
    consumer_tasks.spawn(async move {

     

       
127
       
 
       
        write_store

     

       
128
       
 
       
            .receive_batches(batches)

     

       
129
       
 
       
            .await

     
···

       
131
       
 
       
        Ok(())

     

       
132
       
 
       
    });

     

       
133
       
 
       


     

       
134
       
+
       
    whatever_tasks.spawn(async move {

     

       
135
       
 
       
        do_update_stuff(read_store).await;

     

       
136
       
 
       
        log::warn!("status task ended");

     

       
137
       
 
       
        Ok(())

     
···

       
139
       
 
       


     

       
140
       
 
       
    install_metrics_server()?;

     

       
141
       
 
       


     

       
142
       
+
       
    for (i, t) in consumer_tasks.join_all().await.iter().enumerate() {

     

       
143
       
 
       
        log::warn!("task {i} done: {t:?}");

     

       
144
       
 
       
    }

     

       
145
       
+
       


     

       
146
       
+
       
    println!("consumer tasks all completed, killing the others");

     

       
147
       
+
       
    whatever_tasks.shutdown().await;

     

       
148
       
 
       


     

       
149
       
 
       
    println!("bye!");

     

       
150