Method process_frame

nodedb/src/control/server/sync/session/dispatch.rs:37–196

Process an incoming frame and return a response frame (if any). Security-context parameters are optional — when provided, per-delta RLS enforcement, rate limiting, silent rejection, and DLQ persistence are active. `None` puts the session in permissive mode (testing / internal replication channels). # Timeseries push In production, the listener intercepts `TimeseriesPush` before calling `process

(
        &mut self,
        frame: &SyncFrame,
        jwt_validator: &JwtValidator,
        rls_store: Option<&RlsPolicyStore>,
        audit_log: Option<&mut AuditLog>,
        dlq: Option<&mut Syn

    ///
    /// [`SharedStateDispatcher`]: super::super::timeseries_handler::SharedStateDispatcher
    pub fn process_frame(
        &mut self,
        frame: &SyncFrame,
        jwt_validator: &JwtValidator,
        rls_store: Option<&RlsPolicyStore>,
        audit_log: Option<&mut AuditLog>,
        dlq: Option<&mut SyncDlq>,
        epoch_tracker: Option<&std::sync::Mutex<HashMap<String, u64>>>,
    ) -> Option<SyncFrame> {
        match frame.msg_type {
            SyncMessageType::Handshake => {
                let msg: HandshakeMsg = frame.decode_body()?;
                self.handle_handshake(
                    &msg,
                    jwt_validator,
                    self.server_clock.clone(),
                    epoch_tracker,
                )
            }
            SyncMessageType::DeltaPush => {
                let msg: DeltaPushMsg = frame.decode_body()?;
                self.handle_delta_push(&msg, rls_store, audit_log, dlq)
            }
            SyncMessageType::VectorClockSync => {
                let msg: VectorClockSyncMsg = frame.decode_body()?;
                self.handle_vector_clock_sync(&msg)
            }
            SyncMessageType::ShapeSubscribe => {
                let msg: super::super::shape::handler::ShapeSubscribeMsg = frame.decode_body()?;
                let registry = super::super::shape::registry::ShapeRegistry::new();
                let tenant_id = self.tenant_id.map(|t| t.as_u64()).unwrap_or(0);
                let current_lsn = self.server_clock.values().copied().max().unwrap_or(0);
                // Record the subscription so CollectionPurged broadcast
                // notifies this session when the shape's source
                // collection is hard-deleted. Graph shapes have no
                // single source collection; skip tracking for those.
                if let Some(coll) = msg.shape.collection() {
                    self.track_collection(tenant_id, coll);
                }
                super::super::shape::handler::handle_subscribe(
                    &self.session_id,
                    tenant_id,
                    &msg,
                    &registry,
                    current_lsn,
                    |_shape, _lsn| super::super::shape::handler::ShapeSnapshotData::empty(),
                )
            }
            SyncMessageType::ShapeUnsubscribe => {
                let msg: super::super::shape::handler::ShapeUnsubscribeMsg = frame.decode_body()?;
                let registry = super::super::shape::registry::ShapeRegistry::new();
                super::super::shape::handler::handle_unsubscribe(&self.session_id, &msg, &registry);
                None
            }
            SyncMessageType::TimeseriesPush => {
                // Production path: listener.rs intercepts TimeseriesPush
                // before this dispatch and runs it through
                // SharedStateDispatcher. Reaching this arm means the