udb 0.4.27

Universal Data Broker — a Rust gRPC broker over multiple databases (Postgres, MySQL, SQLite, MongoDB, ClickHouse, Cassandra, MSSQL, Redis, Qdrant, S3, Neo4j, …) with per-tenant RLS, 2PC, sagas, and CDC.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

//! The bounded delta forwarder: the spawned task that filters CDC events against
//! the subscriber's tenant scope (fail closed) and IR predicate, forwarding
//! survivors as `Change` frames over a BOUNDED channel and closing the stream
//! with `resource_exhausted` on broadcast lag or a saturated subscriber.

use std::pin::Pin;

use tokio::sync::{broadcast, mpsc};
use tokio_stream::Stream;
use tonic::Status;

use crate::ir::LogicalFilter;
use crate::proto::udb::core::livequery::services::v1 as lq_pb;

use super::budget::StreamSlot;
use super::errors::livequery_backpressure_status;
use super::predicate::{
    change_frame, change_row, event_matches_tenant_scope, filter_matches_row, topic_matches_source,
};

/// The streamed frame type: an initial snapshot then a stream of change deltas.
pub(crate) type LiveQueryStream =
    Pin<Box<dyn Stream<Item = Result<lq_pb::SubscribeResponse, Status>> + Send + 'static>>;

/// Drive the bounded delta forwarder: subscribe to the CDC broadcast feed, drop
/// every event that fails the fail-closed tenant re-check or the IR predicate,
/// and forward survivors as `Change` frames. Closes the stream with
/// `resource_exhausted` on broadcast lag or a saturated subscriber channel.
pub(crate) async fn run_delta_forward(
    mut rx: broadcast::Receiver<crate::cdc::CdcEnvelope>,
    tx: mpsc::Sender<Result<lq_pb::SubscribeResponse, Status>>,
    tenant_id: String,
    project_id: String,
    cdc_topic: String,
    user_filter: Option<LogicalFilter>,
    // Owned for the lifetime of the live stream; dropping it (on ANY exit path
    // of this task — normal break, error close, abort) releases this
    // subscription's per-tenant active-stream budget slot.
    _stream_slot: StreamSlot,
) {
    loop {
        // Wake on subscriber hang-up too: without `tx.closed()` a disconnected
        // client whose source entity never mutates would park this task (and
        // pin its budget slot) on the broadcast feed indefinitely.
        let received = tokio::select! {
            _ = tx.closed() => break,
            received = rx.recv() => received,
        };
        match received {
            Ok(envelope) => {
                if !topic_matches_source(&envelope.topic, &cdc_topic) {
                    continue;
                }
                // Fail closed if the payload cannot be inspected: an opaque event
                // cannot be proven to belong to this tenant.
                let payload =
                    match serde_json::from_str::<serde_json::Value>(&envelope.payload_json) {
                        Ok(value) => value,
                        Err(_) => continue,
                    };
                // SECURITY: per-event tenant-scope re-check — a tenant-less or
                // foreign event is dropped here, never streamed.
                if !event_matches_tenant_scope(&envelope.topic, &payload, &tenant_id, &project_id) {
                    continue;
                }
                let row = change_row(&payload);
                if let Some(filter) = user_filter.as_ref() {
                    if !filter_matches_row(filter, &row) {
                        continue;
                    }
                }
                match tx.try_send(Ok(change_frame(&envelope, &payload))) {
                    Ok(()) => {}
                    Err(mpsc::error::TrySendError::Full(_)) => {
                        let _ = tx
                            .send(Err(livequery_backpressure_status(
                                "subscriber_channel",
                                "live query subscriber too slow; stream closed",
                            )))
                            .await;
                        break;
                    }
                    Err(mpsc::error::TrySendError::Closed(_)) => break,
                }
            }
            Err(broadcast::error::RecvError::Lagged(_)) => {
                let _ = tx
                    .send(Err(livequery_backpressure_status(
                        "delta feed lag",
                        "live query delta feed lagged; stream closed",
                    )))
                    .await;
                break;
            }
            Err(broadcast::error::RecvError::Closed) => break,
        }
    }
}