oracledb_protocol/thin/
subscr.rs

1#![forbid(unsafe_code)]
2
3//! CQN / Continuous Query Notification wire codecs (sans-io).
4//!
5//! Ports the reference thin subscription messages:
6//! - `impl/thin/messages/subscribe.pyx` — FUNC 125 register/unregister payload
7//!   and the `_process_return_parameters` response decode.
8//! - `impl/thin/messages/notification.pyx` — FUNC 187 NOTIFY payload, the OAC
9//!   record loop (`_process_oac`) and the big-endian inner CQN payload decoder
10//!   (`_process_notification_payload` / `_process_tables` / `_process_rows` /
11//!   `_process_queries`).
12//!
13//! Only the byte<->struct translation lives here; the second ("emon")
14//! connection, the background receive loop and the Python callback invocation
15//! live in the driver and pyshim crates.
16
17use super::*;
18use crate::wire::{ProtocolLimits, TtcReader, TtcWriter};
19
20/// Result of decoding the SUBSCRIBE (register) response.
21#[derive(Clone, Debug, Default, PartialEq, Eq)]
22pub struct SubscribeResult {
23    /// `USER_CHANGE_NOTIFICATION_REGS.REGID` — exposed as `Subscription.id`.
24    pub registration_id: u64,
25    /// EMON client id (e.g. `b"OCI:EP:301"`) echoed back in the NOTIFY message.
26    pub client_id: Option<Vec<u8>>,
27}
28
29/// One row changed inside a table notification.
30#[derive(Clone, Debug, PartialEq, Eq)]
31pub struct MsgRow {
32    pub operation: u32,
33    pub rowid: String,
34}
35
36/// One table changed inside an OBJCHANGE / QUERYCHANGE notification.
37#[derive(Clone, Debug, PartialEq, Eq)]
38pub struct MsgTable {
39    pub operation: u32,
40    pub name: String,
41    pub rows: Vec<MsgRow>,
42}
43
44/// One query whose result set changed (QUERYCHANGE notifications).
45#[derive(Clone, Debug, PartialEq, Eq)]
46pub struct MsgQuery {
47    pub id: u64,
48    pub operation: u32,
49    pub tables: Vec<MsgTable>,
50}
51
52/// A single decoded OAC notification record handed to the user callback.
53#[derive(Clone, Debug, PartialEq, Eq)]
54pub struct NotificationMessage {
55    /// `EVENT_*` value placed on `Message.type`.
56    pub msg_type: u32,
57    pub dbname: Option<String>,
58    /// Thin never decodes the transaction id (14 bytes skipped); always `None`.
59    pub txid: Option<Vec<u8>>,
60    pub registered: bool,
61    pub queue_name: Option<String>,
62    pub consumer_name: Option<String>,
63    pub msgid: Option<Vec<u8>>,
64    pub tables: Vec<MsgTable>,
65    pub queries: Vec<MsgQuery>,
66}
67
68/// Outcome of decoding one OAC record from the notification stream.
69#[derive(Clone, Debug, PartialEq, Eq)]
70pub enum NotificationRecord {
71    /// A record to deliver to the callback. `end_of_response` mirrors the
72    /// reference flag (DEREG / DEREG_NFY terminate the loop after delivery).
73    Message {
74        message: NotificationMessage,
75        end_of_response: bool,
76    },
77    /// `TNS_SUBSCR_STOP_NOTIF` — the stream is finished; no callback fires.
78    Stop,
79}
80
81/// Writes the function-code header plus the pipeline `token_num` (always 0 for
82/// these messages) when the negotiated caps include it, mirroring
83/// `messages/base.pyx::_write_function_code` (writes `ub8 token_num` for
84/// `ttc_field_version >= TNS_CCAP_FIELD_VERSION_23_1_EXT_1`).
85fn write_function_code_token(w: &mut TtcWriter, function_code: u8, seq_num: u8, field_version: u8) {
86    w.write_function_header(function_code, seq_num, field_version);
87}
88
89/// Build the SUBSCRIBE (FUNC 125) payload for register (`opcode = 1`) or
90/// unregister (`opcode = 2`). Ports `subscribe.pyx::_write_message`.
91///
92/// `qos`/`operations` are the *public* `SUBSCR_QOS_*` / `OPCODE_*` values; this
93/// function performs the qos/flags derivation (`subscribe.pyx:82-93`).
94#[allow(clippy::too_many_arguments)]
95pub fn build_subscribe_payload_with_seq(
96    seq_num: u8,
97    opcode: u8,
98    username: Option<&str>,
99    client_id: Option<&[u8]>,
100    namespace: u32,
101    name: Option<&str>,
102    public_qos: u32,
103    operations: u32,
104    timeout: u32,
105    grouping_class: u8,
106    grouping_value: u32,
107    grouping_type: u8,
108    registration_id: u64,
109    field_version: u8,
110) -> Result<Vec<u8>> {
111    // derive the wire qos flags
112    let mut qos = TNS_SUBSCR_QOS_SECURE;
113    if public_qos & SUBSCR_QOS_RELIABLE != 0 {
114        qos |= TNS_SUBSCR_QOS_RELIABLE;
115    }
116    if public_qos & SUBSCR_QOS_DEREG_NFY != 0 {
117        qos |= TNS_SUBSCR_QOS_PURGE_ON_NTFN;
118    }
119    // derive the wire operation flags
120    let mut flags = operations;
121    if public_qos & SUBSCR_QOS_QUERY != 0 {
122        flags |= TNS_SUBSCR_FLAGS_QUERY;
123    }
124    if public_qos & SUBSCR_QOS_ROWIDS != 0 {
125        flags |= TNS_SUBSCR_FLAGS_INCLUDE_ROWIDS;
126    }
127    // grouping_type can only be sent when a grouping class is set
128    let grouping_type = if grouping_class == 0 {
129        0
130    } else {
131        grouping_type
132    };
133
134    let username_bytes = username.map(str::as_bytes);
135
136    let mut w = TtcWriter::new();
137    write_function_code_token(&mut w, TNS_FUNC_SUBSCRIBE, seq_num, field_version);
138    w.write_u8(opcode);
139    w.write_ub4(TNS_SUBSCR_MODE_CLIENT_INITIATED);
140    match username_bytes {
141        Some(bytes) => {
142            w.write_u8(1); // pointer (username)
143            w.write_ub4(u32::try_from(bytes.len()).unwrap_or(u32::MAX));
144        }
145        None => {
146            w.write_u8(0);
147            w.write_ub4(0);
148        }
149    }
150    match client_id {
151        Some(bytes) => {
152            w.write_u8(1); // pointer (location)
153            w.write_ub4(u32::try_from(bytes.len()).unwrap_or(u32::MAX));
154        }
155        None => {
156            w.write_u8(0);
157            w.write_ub4(0);
158        }
159    }
160    w.write_u8(1); // pointer (registration)
161    w.write_ub4(1); // num registrations
162    w.write_ub2(1); // raw presentation
163    w.write_ub2(6); // version for client notification
164    w.write_u8(0); // pointer (namespace out attrs)
165    w.write_u8(1); // pointer (num elements in array)
166    w.write_u8(0); // pointer (generic out attrs)
167    w.write_u8(1); // pointer (num elements in array)
168    if version_gates::writes_subscribe_client_id_block(field_version) {
169        w.write_u8(1); // kpninst
170        w.write_u8(1); // kpninstl
171        w.write_u8(1); // kpngcret
172        w.write_u8(1); // kpngcretl
173        w.write_u8(1); // client id
174        w.write_ub4(TNS_SUBSCR_CLIENT_ID_LEN);
175        w.write_u8(1); // client id length
176    }
177    if let Some(bytes) = username_bytes {
178        w.write_bytes_with_length(bytes)?;
179    }
180    if let Some(bytes) = client_id {
181        w.write_bytes_with_length(bytes)?;
182    }
183    w.write_ub4(namespace);
184    match name {
185        Some(name) => w.write_bytes_with_two_lengths(Some(name.as_bytes()))?,
186        None => w.write_ub4(0),
187    }
188    w.write_ub4(0); // context length
189    w.write_ub4(0); // payload type
190    w.write_ub4(qos);
191    w.write_ub4(0); // payload callback length (JMS)
192    w.write_ub4(timeout);
193    w.write_ub4(0); // kpdnsd
194    w.write_ub4(flags);
195    w.write_ub4(0); // change lag between notifications
196    w.write_ub4(0); // change registration id
197    w.write_u8(grouping_class);
198    w.write_ub4(grouping_value);
199    w.write_u8(grouping_type);
200    w.write_ub4(0); // grouping class start time
201                    // grouping repeat count: write_sb4(0); for the constant 0 this is the same
202                    // single 0x00 byte the unsigned encoder emits.
203    w.write_ub4(0);
204    w.write_ub8(registration_id);
205    Ok(w.into_bytes())
206}
207
208/// Decode the SUBSCRIBE (register) response. Ports
209/// `subscribe.pyx::_process_return_parameters`, dispatched on the
210/// `TNS_MSG_TYPE_PARAMETER` message inside the standard function response loop.
211pub fn parse_subscribe_response(
212    payload: &[u8],
213    capabilities: ClientCapabilities,
214) -> Result<SubscribeResult> {
215    parse_subscribe_response_with_limits(payload, capabilities, ProtocolLimits::DEFAULT)
216}
217
218pub fn parse_subscribe_response_with_limits(
219    payload: &[u8],
220    capabilities: ClientCapabilities,
221    limits: ProtocolLimits,
222) -> Result<SubscribeResult> {
223    let mut reader = TtcReader::with_limits(payload, limits)?;
224    let mut result = SubscribeResult::default();
225    let field_version = capabilities.ttc_field_version;
226    while reader.remaining() > 0 {
227        let message_type = reader.read_u8()?;
228        match message_type {
229            0 => {}
230            TNS_MSG_TYPE_PARAMETER => {
231                parse_subscribe_return_parameters(&mut reader, field_version, &mut result)?;
232            }
233            TNS_MSG_TYPE_STATUS => {
234                let _call_status = reader.read_ub4()?;
235                let _seq = reader.read_ub2()?;
236            }
237            TNS_MSG_TYPE_SERVER_SIDE_PIGGYBACK => {
238                let _ = skip_server_side_piggyback(&mut reader)?;
239            }
240            TNS_MSG_TYPE_END_OF_RESPONSE => break,
241            TNS_MSG_TYPE_ERROR => {
242                let info = parse_server_error_info(&mut reader, field_version)?;
243                if info.number != 0 {
244                    return Err(ProtocolError::ServerError(info.message));
245                }
246            }
247            _ => {
248                return Err(ProtocolError::UnknownMessageType {
249                    message_type,
250                    position: reader.position().saturating_sub(1),
251                })
252            }
253        }
254    }
255    Ok(result)
256}
257
258fn parse_subscribe_return_parameters(
259    reader: &mut TtcReader<'_>,
260    field_version: u8,
261    result: &mut SubscribeResult,
262) -> Result<()> {
263    let num_values = reader.read_ub4()?; // out parameters (kpnrl)
264    for _ in 0..num_values {
265        let _ = reader.read_ub4()?;
266    }
267    for _ in 0..num_values {
268        let _ = reader.read_ub4()?; // registration id (short)
269    }
270    let num_values = reader.read_ub4()?; // out parameters (kpngrl)
271    for _ in 0..num_values {
272        result.registration_id = reader.read_ub8()?;
273        if version_gates::reads_subscribe_response_details(field_version) {
274            let _subscriber_name = reader.read_bytes_with_length()?;
275        }
276    }
277    if version_gates::reads_subscribe_response_details(field_version) {
278        let num_instances = reader.read_ub4()?;
279        for _ in 0..num_instances {
280            let _ = reader.read_bytes_with_length()?;
281        }
282        let num_listeners = reader.read_ub4()?;
283        for _ in 0..num_listeners {
284            let _ = reader.read_bytes_with_length()?;
285        }
286        result.client_id = reader.read_bytes_with_length()?;
287    }
288    Ok(())
289}
290
291/// Build the NOTIFY (FUNC 187) payload sent on the emon connection. Ports
292/// `notification.pyx::_write_message`. The caller must transmit this packet
293/// with the `TNS_DATA_FLAGS_END_OF_REQUEST` data flag set.
294pub fn build_notify_payload_with_seq(
295    seq_num: u8,
296    client_id: &[u8],
297    field_version: u8,
298) -> Result<Vec<u8>> {
299    let mut w = TtcWriter::new();
300    write_function_code_token(&mut w, TNS_FUNC_NOTIFY, seq_num, field_version);
301    w.write_ub4(u32::try_from(client_id.len()).unwrap_or(u32::MAX));
302    w.write_bytes_with_length(client_id)?;
303    w.write_u8(TNS_INIT_KPNDRREQ);
304    w.write_ub4(0);
305    Ok(w.into_bytes())
306}
307
308/// Decode every OAC record in a notification stream. The reference reads one
309/// leading `message_type` byte (`TNS_MSG_TYPE_OAC`) then loops `_process_oac`
310/// until `end_of_response`; the driver chains network packets into `payload`
311/// so this operates on the full concatenated TTC stream.
312///
313/// Returns the decoded records in order. A trailing [`NotificationRecord::Stop`]
314/// (or a record whose `end_of_response` is set) marks the end of the stream.
315pub fn parse_notification_stream(
316    payload: &[u8],
317    namespace: u32,
318    public_qos: u32,
319    db_name: Option<&str>,
320) -> Result<Vec<NotificationRecord>> {
321    parse_notification_stream_with_limits(
322        payload,
323        namespace,
324        public_qos,
325        db_name,
326        ProtocolLimits::DEFAULT,
327    )
328}
329
330pub fn parse_notification_stream_with_limits(
331    payload: &[u8],
332    namespace: u32,
333    public_qos: u32,
334    db_name: Option<&str>,
335    limits: ProtocolLimits,
336) -> Result<Vec<NotificationRecord>> {
337    let mut reader = TtcReader::with_limits(payload, limits)?;
338    let message_type = reader.read_u8()?; // outer process(): read_ub1(message_type)
339    if message_type != TNS_MSG_TYPE_OAC {
340        return Err(ProtocolError::UnknownMessageType {
341            message_type,
342            position: reader.position().saturating_sub(1),
343        });
344    }
345    let mut records = Vec::new();
346    while reader.remaining() > 0 {
347        let record =
348            parse_oac_record_with_limits(&mut reader, namespace, public_qos, db_name, limits)?;
349        let end = match &record {
350            NotificationRecord::Stop => true,
351            NotificationRecord::Message {
352                end_of_response, ..
353            } => *end_of_response,
354        };
355        records.push(record);
356        if end {
357            break;
358        }
359    }
360    Ok(records)
361}
362
363/// Consume the leading `TNS_MSG_TYPE_OAC` byte that precedes the OAC record
364/// stream (`process()` reads it once before delivering any record). Returns the
365/// number of bytes consumed (1) or an error if the byte is not OAC.
366pub fn check_notification_header(bytes: &[u8]) -> Result<usize> {
367    check_notification_header_with_limits(bytes, ProtocolLimits::DEFAULT)
368}
369
370pub fn check_notification_header_with_limits(
371    bytes: &[u8],
372    limits: ProtocolLimits,
373) -> Result<usize> {
374    let mut reader = TtcReader::with_limits(bytes, limits)?;
375    let message_type = reader.read_u8()?;
376    if message_type != TNS_MSG_TYPE_OAC {
377        return Err(ProtocolError::UnknownMessageType {
378            message_type,
379            position: 0,
380        });
381    }
382    Ok(reader.position())
383}
384
385/// Attempt to decode exactly one OAC record from the front of `bytes`. Returns
386/// the decoded record and the number of bytes consumed, or `Ok(None)` when the
387/// buffer does not yet hold a complete record (the caller must read more data
388/// from the EMON socket and retry — mirroring the reference `ReadBuffer`
389/// chaining packets within a single `process()` call).
390pub fn try_parse_oac_record(
391    bytes: &[u8],
392    namespace: u32,
393    public_qos: u32,
394    db_name: Option<&str>,
395) -> Result<Option<(NotificationRecord, usize)>> {
396    try_parse_oac_record_with_limits(
397        bytes,
398        namespace,
399        public_qos,
400        db_name,
401        ProtocolLimits::DEFAULT,
402    )
403}
404
405pub fn try_parse_oac_record_with_limits(
406    bytes: &[u8],
407    namespace: u32,
408    public_qos: u32,
409    db_name: Option<&str>,
410    limits: ProtocolLimits,
411) -> Result<Option<(NotificationRecord, usize)>> {
412    let mut reader = TtcReader::with_limits(bytes, limits)?;
413    match parse_oac_record_with_limits(&mut reader, namespace, public_qos, db_name, limits) {
414        Ok(record) => Ok(Some((record, reader.position()))),
415        // The server only emits well-formed records; a decode failure while the
416        // stream is still being chained means the buffer is short, so signal
417        // "need more bytes" rather than treating it as corruption.
418        Err(_) => Ok(None),
419    }
420}
421
422/// Decode a single OAC record. Ports `notification.pyx::_process_oac` plus the
423/// inner payload decode.
424pub fn parse_oac_record(
425    reader: &mut TtcReader<'_>,
426    namespace: u32,
427    public_qos: u32,
428    db_name: Option<&str>,
429) -> Result<NotificationRecord> {
430    parse_oac_record_with_limits(reader, namespace, public_qos, db_name, reader.limits())
431}
432
433pub fn parse_oac_record_with_limits(
434    reader: &mut TtcReader<'_>,
435    namespace: u32,
436    public_qos: u32,
437    db_name: Option<&str>,
438    limits: ProtocolLimits,
439) -> Result<NotificationRecord> {
440    let message_type = reader.read_ub4()?;
441    if message_type == TNS_SUBSCR_STOP_NOTIF {
442        return Ok(NotificationRecord::Stop);
443    }
444    let _error_code = reader.read_ub4()?;
445    let _registration_id = reader.read_ub4()?;
446    let queue_name = reader.read_string_with_length()?;
447    let consumer_name = reader.read_string_with_length()?;
448    let msgid = reader.read_bytes_with_length()?;
449    let num_props = reader.read_ub4()?;
450    if num_props > 0 {
451        // AQ message properties path: skip the invalid-length byte then the
452        // property records. The CQN tests never exercise this branch (AQ uses
453        // num_props == 0); skip conservatively so the stream stays aligned.
454        let _ = reader.read_u8()?;
455        skip_msg_props(reader, num_props)?;
456    }
457    skip_bytes_with_length(reader)?; // JMS message properties
458
459    let mut payload: Option<Vec<u8>> = None;
460    if namespace != TNS_SUBSCR_NAMESPACE_AQ {
461        let _payload_type = reader.read_ub4()?;
462        let _payload_flags = reader.read_ub4()?;
463        let _chunk_number = reader.read_ub4()?;
464        payload = reader.read_bytes_with_length()?;
465        skip_bytes_with_length(reader)?; // DbObject / JSON payload
466    }
467
468    let mut message = NotificationMessage {
469        msg_type: 0,
470        dbname: db_name.map(str::to_string),
471        txid: None,
472        registered: false,
473        queue_name,
474        consumer_name,
475        msgid,
476        tables: Vec::new(),
477        queries: Vec::new(),
478    };
479    let end_of_response = process_notification_payload(
480        payload.as_deref(),
481        namespace,
482        public_qos,
483        limits,
484        &mut message,
485    )?;
486    Ok(NotificationRecord::Message {
487        message,
488        end_of_response,
489    })
490}
491
492/// Ports `_process_notification_payload`. Returns the resulting
493/// `end_of_response` flag.
494fn process_notification_payload(
495    payload: Option<&[u8]>,
496    namespace: u32,
497    public_qos: u32,
498    limits: ProtocolLimits,
499    message: &mut NotificationMessage,
500) -> Result<bool> {
501    if namespace == TNS_SUBSCR_NAMESPACE_AQ {
502        message.msg_type = EVENT_AQ;
503        return Ok(false);
504    }
505    let Some(payload) = payload else {
506        // empty payload => registration discarded
507        message.msg_type = EVENT_DEREG;
508        return Ok(true);
509    };
510    let mut end_of_response = false;
511    if public_qos & SUBSCR_QOS_DEREG_NFY != 0 {
512        message.registered = false;
513        end_of_response = true;
514    } else {
515        message.registered = true;
516    }
517    // inner payload is a plain big-endian byte cursor
518    let mut cur = ByteCursor::with_limits(payload, limits)?;
519    let _version = cur.u16be()?;
520    let _registration_id = cur.u32be()?;
521    let event_type = cur.u32be()?;
522    message.msg_type = event_type;
523    let dbname_len = cur.u16be()? as usize;
524    let dbname = cur.raw(dbname_len)?;
525    message.dbname = Some(
526        String::from_utf8(dbname.to_vec())
527            .map_err(|_| ProtocolError::TtcDecode("notification dbname not UTF-8"))?,
528    );
529    cur.skip(14)?; // transaction id + SCN (txid intentionally left None)
530    if event_type == EVENT_OBJCHANGE {
531        message.tables = process_tables(&mut cur)?;
532    } else if event_type == EVENT_QUERYCHANGE {
533        message.queries = process_queries(&mut cur)?;
534    }
535    Ok(end_of_response)
536}
537
538fn process_tables(cur: &mut ByteCursor<'_>) -> Result<Vec<MsgTable>> {
539    let num_tables = cur.u16be()?;
540    // Each table record reads at least a u32 operation + u16 name length (6
541    // bytes) before its name, so cap the reservation by the buffer
542    // (BoundedReader); the loop still fails closed on truncation.
543    let mut tables: Vec<MsgTable> = cur.with_capacity_limited(
544        num_tables as usize,
545        6,
546        ProtocolLimits::check_length_prefixed_elements,
547    )?;
548    for _ in 0..num_tables {
549        let operation = cur.u32be()?;
550        let name_len = cur.u16be()? as usize;
551        let name = String::from_utf8(cur.raw(name_len)?.to_vec())
552            .map_err(|_| ProtocolError::TtcDecode("table name not UTF-8"))?;
553        let _object_num = cur.u32be()?;
554        let rows = if operation & OPCODE_ALLROWS == 0 {
555            process_rows(cur)?
556        } else {
557            Vec::new()
558        };
559        tables.push(MsgTable {
560            operation,
561            name,
562            rows,
563        });
564    }
565    Ok(tables)
566}
567
568fn process_rows(cur: &mut ByteCursor<'_>) -> Result<Vec<MsgRow>> {
569    let num_rows = cur.u16be()?;
570    // Each row record reads at least a u32 operation + u16 rowid length (6
571    // bytes); bound the reservation by the buffer (BoundedReader).
572    let mut rows: Vec<MsgRow> = cur.with_capacity_limited(
573        num_rows as usize,
574        6,
575        ProtocolLimits::check_length_prefixed_elements,
576    )?;
577    for _ in 0..num_rows {
578        let operation = cur.u32be()?;
579        let rowid_len = cur.u16be()? as usize;
580        let rowid = String::from_utf8(cur.raw(rowid_len)?.to_vec())
581            .map_err(|_| ProtocolError::TtcDecode("rowid not UTF-8"))?;
582        rows.push(MsgRow { operation, rowid });
583    }
584    Ok(rows)
585}
586
587fn process_queries(cur: &mut ByteCursor<'_>) -> Result<Vec<MsgQuery>> {
588    let num_queries = cur.u16be()?;
589    // Each query record reads at least three u32s (12 bytes) before its nested
590    // tables; bound the reservation by the buffer (BoundedReader).
591    let mut queries: Vec<MsgQuery> = cur.with_capacity_limited(
592        num_queries as usize,
593        12,
594        ProtocolLimits::check_length_prefixed_elements,
595    )?;
596    for _ in 0..num_queries {
597        let id_lsb = u64::from(cur.u32be()?);
598        let id_msb = u64::from(cur.u32be()?);
599        let id = (id_msb << 32) | id_lsb;
600        let operation = cur.u32be()?;
601        let tables = process_tables(cur)?;
602        queries.push(MsgQuery {
603            id,
604            operation,
605            tables,
606        });
607    }
608    Ok(queries)
609}
610
611/// Skip AQ message-property records (`_process_msg_props`). The CQN tests never
612/// reach this branch; this keeps the parser aligned should the server send it.
613fn skip_msg_props(reader: &mut TtcReader<'_>, num_props: u32) -> Result<()> {
614    for _ in 0..num_props {
615        skip_bytes_with_length(reader)?; // name
616        skip_bytes_with_length(reader)?; // value
617    }
618    Ok(())
619}
620
621fn skip_bytes_with_length(reader: &mut TtcReader<'_>) -> Result<()> {
622    let _ = reader.read_bytes_with_length()?;
623    Ok(())
624}
625
626/// A plain big-endian cursor over the inner CQN payload bytes (no TTC chunking).
627struct ByteCursor<'a> {
628    bytes: &'a [u8],
629    pos: usize,
630    limits: ProtocolLimits,
631}
632
633impl<'a> ByteCursor<'a> {
634    #[cfg(test)]
635    fn new(bytes: &'a [u8]) -> Self {
636        Self {
637            bytes,
638            pos: 0,
639            limits: ProtocolLimits::DEFAULT,
640        }
641    }
642
643    fn with_limits(bytes: &'a [u8], limits: ProtocolLimits) -> Result<Self> {
644        let limits = limits.validate()?;
645        limits.check_response_bytes(bytes.len())?;
646        Ok(Self {
647            bytes,
648            pos: 0,
649            limits,
650        })
651    }
652
653    fn raw(&mut self, n: usize) -> Result<&'a [u8]> {
654        let end = self
655            .pos
656            .checked_add(n)
657            .ok_or(ProtocolError::TtcDecode("notification payload overflow"))?;
658        let slice = self
659            .bytes
660            .get(self.pos..end)
661            .ok_or(ProtocolError::TtcDecode("notification payload truncated"))?;
662        self.pos = end;
663        Ok(slice)
664    }
665
666    fn skip(&mut self, n: usize) -> Result<()> {
667        let _ = self.raw(n)?;
668        Ok(())
669    }
670
671    fn u16be(&mut self) -> Result<u16> {
672        let bytes = self.raw(2)?;
673        Ok(u16::from_be_bytes([bytes[0], bytes[1]]))
674    }
675
676    fn u32be(&mut self) -> Result<u32> {
677        let bytes = self.raw(4)?;
678        Ok(u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]))
679    }
680}
681
682impl crate::wire::BoundedReader for ByteCursor<'_> {
683    fn remaining(&self) -> usize {
684        self.bytes.len().saturating_sub(self.pos)
685    }
686
687    fn protocol_limits(&self) -> ProtocolLimits {
688        self.limits
689    }
690}
691
692#[cfg(test)]
693mod tests {
694    use super::*;
695
696    // BoundedReader invariant (l2p), CQN array family: a notification table
697    // record declaring the maximum num_tables (0xFFFF) but carrying no table
698    // bytes must fail closed via the bounded reservation + the per-record read,
699    // not pre-allocate 65535 MsgTable structs from the count. (num_tables is a
700    // u16 so this was never a multi-GB OOM, but routing it through the bound
701    // keeps the whole class uniform and regression-proof.)
702    #[test]
703    fn cqn_oversized_table_count_fails_closed_not_oom() {
704        // num_tables = 0xFFFF (u16), then nothing.
705        let bytes = [0xFFu8, 0xFF];
706        let mut cur = ByteCursor::new(&bytes);
707        let err = process_tables(&mut cur).expect_err("oversized table count must fail closed");
708        assert!(matches!(err, ProtocolError::TtcDecode(_)), "got {err:?}");
709        // The pre-allocation never exceeds remaining()/6 even for the max count.
710        let cur2 = ByteCursor::new(&bytes);
711        let v: Vec<MsgTable> = cur2.with_capacity_bounded(0xFFFF, 6);
712        assert!(v.capacity() <= 1, "reservation capped by remaining bytes");
713    }
714
715    #[test]
716    fn cqn_table_count_respects_protocol_element_limit() {
717        // num_tables = 2. A max_length_prefixed_elements=1 policy rejects the
718        // count before reserving table slots.
719        let bytes = [0x00u8, 0x02];
720        let limits = ProtocolLimits {
721            max_length_prefixed_elements: 1,
722            ..ProtocolLimits::DEFAULT
723        };
724        let mut cur = ByteCursor::with_limits(&bytes, limits).expect("valid limits");
725        let err = process_tables(&mut cur).expect_err("table count above policy must fail");
726        assert!(
727            matches!(
728                err,
729                ProtocolError::ResourceLimit {
730                    limit: "length_prefixed_elements",
731                    observed: 2,
732                    maximum: 1,
733                }
734            ),
735            "got {err:?}"
736        );
737    }
738
739    fn caps_12_1() -> ClientCapabilities {
740        ClientCapabilities {
741            ttc_field_version: 24,
742            ..ClientCapabilities::default()
743        }
744    }
745
746    #[test]
747    fn subscribe_register_payload_matches_golden() {
748        // Golden /tmp/cqn_trace.txt line 2421 (op 8, socket 5), payload after
749        // the 2-byte data flags. seq byte is 0x03 in the capture.
750        let payload = build_subscribe_payload_with_seq(
751            0x03,
752            TNS_SUBSCR_OP_REGISTER,
753            Some("pythontest"),
754            None,
755            TNS_SUBSCR_NAMESPACE_DBCHANGE,
756            None,
757            SUBSCR_QOS_ROWIDS,
758            0, // OPCODE_ALLOPS
759            10,
760            0,
761            0,
762            0,
763            0,
764            24,
765        )
766        .expect("subscribe payload");
767        // real capture TTC payload (token byte 0x00 follows the seq):
768        // 03 7d 03 00 01 01 04 01 01 0a 00 00 01 01 01 01 01 01 06 00 ...
769        let expected: &[u8] = &[
770            0x03, 0x7D, 0x03, 0x00, 0x01, 0x01, 0x04, 0x01, 0x01, 0x0A, 0x00, 0x00, 0x01, 0x01,
771            0x01, 0x01, 0x01, 0x01, 0x06, 0x00, 0x01, 0x00, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01,
772            0x01, 0x1D, 0x01, 0x0A, 0x70, 0x79, 0x74, 0x68, 0x6F, 0x6E, 0x74, 0x65, 0x73, 0x74,
773            0x01, 0x02, 0x00, 0x00, 0x00, 0x01, 0x08, 0x00, 0x01, 0x0A, 0x00, 0x01, 0x10, 0x00,
774            0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
775        ];
776        assert_eq!(payload, expected);
777    }
778
779    #[test]
780    fn subscribe_unregister_payload_matches_golden() {
781        // Golden /tmp/cqn_trace.txt line 4029 (op 22, socket 5). seq 0x0A,
782        // opcode 2, client_id now set to "OCI:EP:301", reg id 302 in the tail.
783        let payload = build_subscribe_payload_with_seq(
784            0x0A,
785            TNS_SUBSCR_OP_UNREGISTER,
786            Some("pythontest"),
787            Some(b"OCI:EP:301"),
788            TNS_SUBSCR_NAMESPACE_DBCHANGE,
789            None,
790            SUBSCR_QOS_ROWIDS,
791            0,
792            10,
793            0,
794            0,
795            0,
796            302,
797            24,
798        )
799        .expect("unsubscribe payload");
800        let expected: &[u8] = &[
801            0x03, 0x7D, 0x0A, 0x00, 0x02, 0x01, 0x04, 0x01, 0x01, 0x0A, 0x01, 0x01, 0x0A, 0x01,
802            0x01, 0x01, 0x01, 0x01, 0x01, 0x06, 0x00, 0x01, 0x00, 0x01, 0x01, 0x01, 0x01, 0x01,
803            0x01, 0x01, 0x1D, 0x01, 0x0A, 0x70, 0x79, 0x74, 0x68, 0x6F, 0x6E, 0x74, 0x65, 0x73,
804            0x74, 0x0A, 0x4F, 0x43, 0x49, 0x3A, 0x45, 0x50, 0x3A, 0x33, 0x30, 0x31, 0x01, 0x02,
805            0x00, 0x00, 0x00, 0x01, 0x08, 0x00, 0x01, 0x0A, 0x00, 0x01, 0x10, 0x00, 0x00, 0x00,
806            0x00, 0x00, 0x00, 0x00, 0x02, 0x01, 0x2E,
807        ];
808        assert_eq!(payload, expected);
809    }
810
811    #[test]
812    fn notify_payload_matches_golden() {
813        // Golden /tmp/cqn_trace.txt line 3647 (op 8, socket 6) after data flags.
814        let payload =
815            build_notify_payload_with_seq(0x03, b"OCI:EP:301", 24).expect("notify payload");
816        // 03 bb 03 00 01 0a 0a OCI:EP:301 01 00  (token 0x00 after seq)
817        let want: &[u8] = &[
818            0x03, 0xBB, 0x03, 0x00, 0x01, 0x0A, 0x0A, 0x4F, 0x43, 0x49, 0x3A, 0x45, 0x50, 0x3A,
819            0x33, 0x30, 0x31, 0x01, 0x00,
820        ];
821        assert_eq!(payload, want);
822    }
823
824    #[test]
825    fn subscribe_response_decodes_registration_and_client_id() {
826        // Golden /tmp/cqn_trace.txt line 2433 (op 9, socket 5) after data flags.
827        let payload: &[u8] = &[
828            0x08, 0x01, 0x01, 0x00, 0x02, 0x01, 0x2E, 0x01, 0x01, 0x02, 0x01, 0x2E, 0x00, 0x00,
829            0x01, 0x01, 0x01, 0x36, 0x36, 0x28, 0x41, 0x44, 0x44, 0x52, 0x45, 0x53, 0x53, 0x3D,
830            0x28, 0x50, 0x52, 0x4F, 0x54, 0x4F, 0x43, 0x4F, 0x4C, 0x3D, 0x54, 0x43, 0x50, 0x29,
831            0x28, 0x48, 0x4F, 0x53, 0x54, 0x3D, 0x32, 0x39, 0x30, 0x61, 0x63, 0x30, 0x33, 0x30,
832            0x30, 0x33, 0x38, 0x37, 0x29, 0x28, 0x50, 0x4F, 0x52, 0x54, 0x3D, 0x31, 0x35, 0x32,
833            0x31, 0x29, 0x29, 0x01, 0x0A, 0x0A, 0x4F, 0x43, 0x49, 0x3A, 0x45, 0x50, 0x3A, 0x33,
834            0x30, 0x31, 0x09, 0x01, 0x01, 0x02, 0xDD, 0x48, 0x1D,
835        ];
836        let result = parse_subscribe_response(payload, caps_12_1()).expect("subscribe response");
837        assert_eq!(result.registration_id, 302);
838        assert_eq!(result.client_id.as_deref(), Some(&b"OCI:EP:301"[..]));
839    }
840
841    /// The full real notification stream captured on the emon socket
842    /// (`/tmp/cqn_notif_stream.bin`): the leading OAC ack byte plus five OAC
843    /// records (insert / update / insert / delete / truncate).
844    const NOTIF_STREAM: &[u8] = &[
845        0x0d, 0x01, 0x03, 0x00, 0x02, 0x01, 0x2e, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x01, 0x00,
846        0x00, 0x01, 0x60, 0x60, 0x00, 0x01, 0x02, 0xa4, 0xe2, 0x7a, 0x00, 0x00, 0x00, 0x06, 0x00,
847        0x08, 0x46, 0x52, 0x45, 0x45, 0x50, 0x44, 0x42, 0x31, 0x01, 0x00, 0x10, 0x00, 0xd2, 0x03,
848        0x00, 0x00, 0xe2, 0x7a, 0x00, 0x00, 0x00, 0x9b, 0x00, 0x01, 0x00, 0x00, 0x00, 0x02, 0x00,
849        0x18, 0x50, 0x59, 0x54, 0x48, 0x4f, 0x4e, 0x54, 0x45, 0x53, 0x54, 0x2e, 0x54, 0x45, 0x53,
850        0x54, 0x54, 0x45, 0x4d, 0x50, 0x54, 0x41, 0x42, 0x4c, 0x45, 0x00, 0x01, 0x1c, 0x4a, 0x00,
851        0x01, 0x00, 0x00, 0x00, 0x02, 0x00, 0x12, 0x41, 0x41, 0x41, 0x53, 0x6a, 0x4d, 0x41, 0x41,
852        0x59, 0x41, 0x41, 0x41, 0x4a, 0x4f, 0x33, 0x41, 0x41, 0x41, 0x00, 0x01, 0x03, 0x00, 0x02,
853        0x01, 0x2e, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x01, 0x00, 0x00, 0x01, 0x60, 0x60, 0x00,
854        0x01, 0x00, 0x00, 0x89, 0x00, 0x00, 0x00, 0x00, 0x06, 0x00, 0x08, 0x46, 0x52, 0x45, 0x45,
855        0x50, 0x44, 0x42, 0x31, 0x03, 0x00, 0x19, 0x00, 0x98, 0x04, 0x00, 0x00, 0x0b, 0x00, 0x00,
856        0x00, 0x01, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x04, 0x00, 0x18, 0x50, 0x59, 0x54, 0x48,
857        0x4f, 0x4e, 0x54, 0x45, 0x53, 0x54, 0x2e, 0x54, 0x45, 0x53, 0x54, 0x54, 0x45, 0x4d, 0x50,
858        0x54, 0x41, 0x42, 0x4c, 0x45, 0x00, 0x01, 0x1c, 0x4a, 0x00, 0x01, 0x00, 0x00, 0x00, 0x04,
859        0x00, 0x12, 0x41, 0x41, 0x41, 0x53, 0x6a, 0x4d, 0x41, 0x41, 0x59, 0x41, 0x41, 0x41, 0x4a,
860        0x4f, 0x33, 0x41, 0x41, 0x41, 0x00, 0x01, 0x03, 0x00, 0x02, 0x01, 0x2e, 0x00, 0x00, 0x00,
861        0x00, 0x00, 0x01, 0x01, 0x00, 0x00, 0x01, 0x60, 0x60, 0x00, 0x01, 0x03, 0x00, 0x89, 0x00,
862        0x00, 0x00, 0x00, 0x06, 0x00, 0x08, 0x46, 0x52, 0x45, 0x45, 0x50, 0x44, 0x42, 0x31, 0x05,
863        0x00, 0x06, 0x00, 0xa9, 0x04, 0x00, 0x00, 0xe2, 0x7a, 0x00, 0x00, 0x44, 0x32, 0x00, 0x01,
864        0x00, 0x00, 0x00, 0x02, 0x00, 0x18, 0x50, 0x59, 0x54, 0x48, 0x4f, 0x4e, 0x54, 0x45, 0x53,
865        0x54, 0x2e, 0x54, 0x45, 0x53, 0x54, 0x54, 0x45, 0x4d, 0x50, 0x54, 0x41, 0x42, 0x4c, 0x45,
866        0x00, 0x01, 0x1c, 0x4a, 0x00, 0x01, 0x00, 0x00, 0x00, 0x02, 0x00, 0x12, 0x41, 0x41, 0x41,
867        0x53, 0x6a, 0x4d, 0x41, 0x41, 0x59, 0x41, 0x41, 0x41, 0x4a, 0x4f, 0x33, 0x41, 0x41, 0x42,
868        0x00, 0x01, 0x03, 0x00, 0x02, 0x01, 0x2e, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x01, 0x00,
869        0x00, 0x01, 0x60, 0x60, 0x00, 0x01, 0x03, 0xa5, 0xe2, 0x7a, 0x00, 0x00, 0x00, 0x06, 0x00,
870        0x08, 0x46, 0x52, 0x45, 0x45, 0x50, 0x44, 0x42, 0x31, 0x02, 0x00, 0x09, 0x00, 0x7d, 0x04,
871        0x00, 0x00, 0xe2, 0x7a, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x08, 0x00,
872        0x18, 0x50, 0x59, 0x54, 0x48, 0x4f, 0x4e, 0x54, 0x45, 0x53, 0x54, 0x2e, 0x54, 0x45, 0x53,
873        0x54, 0x54, 0x45, 0x4d, 0x50, 0x54, 0x41, 0x42, 0x4c, 0x45, 0x00, 0x01, 0x1c, 0x4a, 0x00,
874        0x01, 0x00, 0x00, 0x00, 0x08, 0x00, 0x12, 0x41, 0x41, 0x41, 0x53, 0x6a, 0x4d, 0x41, 0x41,
875        0x59, 0x41, 0x41, 0x41, 0x4a, 0x4f, 0x33, 0x41, 0x41, 0x42, 0x00, 0x01, 0x03, 0x00, 0x02,
876        0x01, 0x2e, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x01, 0x00, 0x00, 0x01, 0x46, 0x46, 0x00,
877        0x01, 0x00, 0x00, 0x89, 0x00, 0x00, 0x00, 0x00, 0x06, 0x00, 0x08, 0x46, 0x52, 0x45, 0x45,
878        0x50, 0x44, 0x42, 0x31, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xfe, 0x7f, 0x00,
879        0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x11, 0x00, 0x18, 0x50, 0x59, 0x54, 0x48,
880        0x4f, 0x4e, 0x54, 0x45, 0x53, 0x54, 0x2e, 0x54, 0x45, 0x53, 0x54, 0x54, 0x45, 0x4d, 0x50,
881        0x54, 0x41, 0x42, 0x4c, 0x45, 0x00, 0x01, 0x1c, 0x4a, 0x00,
882    ];
883
884    #[test]
885    fn notification_stream_decodes_dml_events() {
886        let records = parse_notification_stream(
887            NOTIF_STREAM,
888            TNS_SUBSCR_NAMESPACE_DBCHANGE,
889            SUBSCR_QOS_ROWIDS,
890            Some("FREEPDB1"),
891        )
892        .expect("notification stream");
893        let messages: Vec<&NotificationMessage> = records
894            .iter()
895            .filter_map(|r| match r {
896                NotificationRecord::Message { message, .. } => Some(message),
897                NotificationRecord::Stop => None,
898            })
899            .collect();
900        assert_eq!(messages.len(), 5);
901
902        let table_ops: Vec<u32> = messages.iter().map(|m| m.tables[0].operation).collect();
903        assert_eq!(table_ops, vec![2, 4, 2, 8, 17]);
904
905        let mut row_ops = Vec::new();
906        let mut rowids = Vec::new();
907        for m in &messages {
908            assert_eq!(m.msg_type, EVENT_OBJCHANGE);
909            assert_eq!(m.dbname.as_deref(), Some("FREEPDB1"));
910            assert!(m.registered);
911            assert!(m.txid.is_none());
912            for row in &m.tables[0].rows {
913                row_ops.push(row.operation);
914                rowids.push(row.rowid.clone());
915            }
916        }
917        assert_eq!(row_ops, vec![2, 4, 2, 8]);
918        assert_eq!(
919            rowids,
920            vec![
921                "AAASjMAAYAAAJO3AAA",
922                "AAASjMAAYAAAJO3AAA",
923                "AAASjMAAYAAAJO3AAB",
924                "AAASjMAAYAAAJO3AAB",
925            ]
926        );
927        // the truncate record carries the ALLROWS bit, so no rows are present
928        assert!(messages[4].tables[0].rows.is_empty());
929    }
930
931    // ---- 12.1 version-gate boundary tests ---------------------------------
932    //
933    // Reference messages/subscribe.pyx gates the client-id pointer block on the
934    // write side (:127) and the subscriber name + RAC instance/listener block
935    // on the read side (:61, :63), all on ttc field version >= 12.1 (7). 12.1's
936    // field version (7) is far below our live floor (18c == 11), so no live lane
937    // ever exercises the pre-12.1 branch; these offline tests pin both sides.
938
939    fn assert_single_insertion(lo: &[u8], hi: &[u8], label: &str) {
940        assert!(hi.len() > lo.len(), "{label}: gated block must add bytes");
941        let prefix = lo.iter().zip(hi).take_while(|(a, b)| a == b).count();
942        let suffix = lo[prefix..]
943            .iter()
944            .rev()
945            .zip(hi[prefix..].iter().rev())
946            .take_while(|(a, b)| a == b)
947            .count();
948        assert_eq!(
949            prefix + suffix,
950            lo.len(),
951            "{label}: below-boundary bytes must equal above-boundary bytes minus one inserted block"
952        );
953    }
954
955    // Reference messages/subscribe.pyx:127 — the kpninst/client-id pointer block.
956    #[test]
957    fn subscribe_build_gates_client_id_block_on_12_1() {
958        let build = |fv| {
959            build_subscribe_payload_with_seq(
960                1,
961                TNS_SUBSCR_OP_REGISTER,
962                None,
963                None,
964                TNS_SUBSCR_NAMESPACE_DBCHANGE,
965                None,
966                0,
967                0,
968                0,
969                0,
970                0,
971                0,
972                0,
973                fv,
974            )
975            .expect("subscribe payload")
976        };
977        assert_single_insertion(
978            &build(TNS_CCAP_FIELD_VERSION_12_1 - 1),
979            &build(TNS_CCAP_FIELD_VERSION_12_1),
980            "subscribe client-id block (12.1)",
981        );
982    }
983
984    // Reference messages/subscribe.pyx:61,63 — subscriber name + RAC instance
985    // and listener addresses, read only at >= 12.1.
986    #[test]
987    fn subscribe_response_read_gates_subscriber_and_instances_on_12_1() {
988        let fixture = |fv: u8| {
989            let mut w = TtcWriter::new();
990            w.write_ub4(0); // kpnrl count (short registration ids)
991            w.write_ub4(1); // kpngrl count (one registration)
992            w.write_ub8(302); // registration id
993            if fv >= TNS_CCAP_FIELD_VERSION_12_1 {
994                w.write_bytes_with_two_lengths(Some(b"SUB"))
995                    .expect("subscriber name");
996            }
997            if fv >= TNS_CCAP_FIELD_VERSION_12_1 {
998                w.write_ub4(0); // num instances
999                w.write_ub4(0); // num listeners
1000                w.write_bytes_with_two_lengths(Some(b"OCI:EP:301"))
1001                    .expect("client id");
1002            }
1003            w.into_bytes()
1004        };
1005        let parse = |bytes: &[u8], fv: u8| -> Option<(u64, Option<Vec<u8>>, usize)> {
1006            let mut reader = TtcReader::new(bytes);
1007            let mut result = SubscribeResult::default();
1008            parse_subscribe_return_parameters(&mut reader, fv, &mut result)
1009                .ok()
1010                .map(|()| (result.registration_id, result.client_id, reader.remaining()))
1011        };
1012        let lo = TNS_CCAP_FIELD_VERSION_12_1 - 1;
1013        let hi = TNS_CCAP_FIELD_VERSION_12_1;
1014
1015        // Version-matched parses consume the record exactly.
1016        assert_eq!(parse(&fixture(lo), lo), Some((302, None, 0)));
1017        assert_eq!(
1018            parse(&fixture(hi), hi),
1019            Some((302, Some(b"OCI:EP:301".to_vec()), 0))
1020        );
1021
1022        // The >= 12.1 block present but read as pre-12.1: the client id is never
1023        // read and the block is left unconsumed on the wire.
1024        let (reg, client, remaining) = parse(&fixture(hi), lo).expect("reg id still parses");
1025        assert_eq!(reg, 302);
1026        assert_eq!(client, None, "pre-12.1 read must not consume the client id");
1027        assert!(
1028            remaining > 0,
1029            "pre-12.1 read leaves the 12.1 block unconsumed"
1030        );
1031
1032        // The block absent but read as >= 12.1: the parser expects a subscriber
1033        // name that is not there and fails closed.
1034        assert_eq!(parse(&fixture(lo), hi), None, "over-read must fail closed");
1035    }
1036}
oracledb_protocol/thin/subscr.rs

oracledb_protocol/thin/
subscr.rs
