fix(qconnect): remove wrong inner envelope, decode QConnectBatch directly

Root cause: inner_payload from CloudPayload.payload IS a QConnectBatch
(QConnectMessages) with repeated QConnectMessage entries in field 3.

The QCloudInnerEnvelope struct incorrectly treated field 3 as opaque
bytes, keeping only the last repeated entry and losing all messages.
Additionally, QConnectMessages.messages_time was declared as uint64
(wire type 0/varint) but the server sends fixed64 (wire type 1),
causing a wire type mismatch that would fail decoding.

Changes:
- Remove QCloudInnerEnvelope struct entirely
- Pass inner_payload directly to decode_queue_server_events/decode_renderer_server_commands
- Fix messages_time from uint64 to fixed64 to match server wire format

Verified against qonductor proto definitions (QConnectBatch) used in
working POC v1 and v2 implementations.
This commit is contained in:
vicrodh committed 2026-02-21 17:15:34 -06:00
1 parent 975fac55f7
commit 4c7be7597d
2 files changed
+13 -41

No files matched your search

+12 -40
View File
@@ -594,33 +594,17 @@ fn handle_incoming_binary(
},
);
// The server wraps QConnectMessages inside an inner envelope with:
// field 1 (fixed64): correlation ID
// field 2 (varint): sequence number
// field 3 (bytes): the actual QConnectMessage protobuf
// We must unwrap field 3 before passing to the decoders.
let qconnect_bytes = match QCloudInnerEnvelope::decode(inner_payload.as_slice()) {
Ok(envelope) => {
let msg = envelope.message.unwrap_or_default();
log::info!(
"[QConnect/Decode] Envelope unwrapped: corr={:?} seq={:?} message_len={}",
envelope.correlation_id,
envelope.sequence,
msg.len()
);
msg
}
Err(err) => {
log::warn!(
"[QConnect/Decode] Envelope decode failed ({}), using raw {} bytes",
err,
inner_payload.len()
);
inner_payload
}
};
// inner_payload is a QConnectBatch (QConnectMessages) protobuf:
// field 1 (fixed64): messages_time
// field 2 (int32): messages_id
// field 3 (repeated): QConnectMessage entries
// Pass directly to decoders — no inner envelope unwrapping needed.
log::info!(
"[QConnect/Decode] Decoding QConnectBatch: {} bytes",
inner_payload.len()
);
match decode_queue_server_events(&qconnect_bytes) {
match decode_queue_server_events(&inner_payload) {
Ok(events) => {
log::info!("[QConnect/Decode] Queue events decoded: {}", events.len());
for event in events {
@@ -632,7 +616,7 @@ fn handle_incoming_binary(
}
}
match decode_renderer_server_commands(&qconnect_bytes) {
match decode_renderer_server_commands(&inner_payload) {
Ok(commands) => {
if !commands.is_empty() {
log::info!("[QConnect/Decode] Renderer commands decoded: {}", commands.len());
@@ -649,7 +633,7 @@ fn handle_incoming_binary(
}
}
if let Ok(inbound) = decode_inbound_json(&qconnect_bytes) {
if let Ok(inbound) = decode_inbound_json(&inner_payload) {
emit(events_tx, TransportEvent::InboundReceived(inbound));
}
}
@@ -832,18 +816,6 @@ struct CloudPayload {
pub payload: Option<Vec<u8>>,
}
/// Inner envelope that wraps QConnectMessage inside CloudPayload.payload.
/// The server sends: CloudPayload { payload: QCloudInnerEnvelope { message: QConnectMessage } }
#[derive(Clone, PartialEq, ::prost::Message)]
struct QCloudInnerEnvelope {
#[prost(fixed64, optional, tag = "1")]
pub correlation_id: Option<u64>,
#[prost(uint64, optional, tag = "2")]
pub sequence: Option<u64>,
#[prost(bytes = "vec", optional, tag = "3")]
pub message: Option<Vec<u8>>,
}
#[cfg(test)]
mod tests {
use super::*;