// Setup PgNotify listener for relevant streams.
        let mut listener = PgListener::connect_with(pool).await.unwrap();
        listener.listen("PARAM_INTERMEDIATES").await.unwrap();
        listener.listen("PARAM_LATEST").await.unwrap();

        loop {
            tokio::select! {
                biased;

                res = self.listener.try_recv() => {
                    let notification = res.unwrap().unwrap();
                    match notification.channel() {
                        "PARAM_INTERMEDIATES" => self.on_intermediary_param(notification),
                        "PARAM_LATEST" => self.on_param(notification),
                        _ => panic!("Unexpected notification; notification={notification:?}")
                    }
                }

                opt = self.worker_handles.next() => {
                    let (name, res) = opt.unwrap();

                    res.unwrap();

                    panic!("Worker exited without panic; worker={name}");
                }

                _ = self.snapshot_interval.tick() => {
                    self.pub_tx.try_send((
                        PUBLISHER_TOPIC,
                        ParamMsg::Snapshot(self.params.clone())
                    )).unwrap();
                }

                _ = self.cxl.cancelled() => break,
            }
        }