#2 Tighten reply correlation: match (sys_bytes, stream, function) exactly
The previous heuristic ("function % 2 == 0 && pending_requests_.count(sys)")
worked in practice but was wrong in principle — SECS-II doesn't enforce
function parity, and a peer protocol violation (replying with the wrong
SxFy) would have been silently treated as a primary message.
Now PendingRequest carries the expected reply stream + function (computed
from request.stream / request.function+1 per SECS-II convention) at
send_request time. handle_data matches on all three:
it->second.expected_stream == h.stream() &&
it->second.expected_function == h.function()
If sys_bytes matches but stream/function doesn't, the Connection logs
a diagnostic ("!! unexpected SxFy for pending sys=N (expected ...)")
and treats the message as a primary so the application handler can
still respond. The pending request stays open until T3.
No behaviour change on the happy path; the demo and all 69 tests still
pass.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -108,8 +108,12 @@ class Connection : public std::enable_shared_from_this<Connection> {
|
||||
|
||||
uint32_t next_system_bytes_ = 1;
|
||||
|
||||
// outstanding data request transactions, keyed by system bytes
|
||||
// outstanding data request transactions, keyed by system bytes.
|
||||
// We track the EXPECTED reply stream/function so we can match exactly
|
||||
// instead of inferring "this is a reply" from function parity.
|
||||
struct PendingRequest {
|
||||
uint8_t expected_stream;
|
||||
uint8_t expected_function;
|
||||
ReplyHandler cb;
|
||||
std::shared_ptr<asio::steady_timer> t3;
|
||||
};
|
||||
|
||||
+24
-4
@@ -113,10 +113,14 @@ void Connection::handle_frame(Frame frame) {
|
||||
void Connection::handle_data(const Frame& frame) {
|
||||
const Header& h = frame.header;
|
||||
|
||||
// Reply correlation: an even function with matching system bytes completes an
|
||||
// outstanding request.
|
||||
// Reply correlation: match on (system_bytes, stream, function) exactly.
|
||||
// SECS-II replies use the same system_bytes as their request and have
|
||||
// stream==request.stream, function==request.function+1; we recorded both
|
||||
// when send_request stored the pending transaction.
|
||||
auto it = pending_requests_.find(h.system_bytes);
|
||||
if ((h.function() % 2 == 0) && it != pending_requests_.end()) {
|
||||
if (it != pending_requests_.end() &&
|
||||
it->second.expected_stream == h.stream() &&
|
||||
it->second.expected_function == h.function()) {
|
||||
secs2::Message reply;
|
||||
try {
|
||||
reply = secs2::Message::from_body(h.stream(), h.function(), h.w_bit(), frame.body);
|
||||
@@ -132,6 +136,18 @@ void Connection::handle_data(const Frame& frame) {
|
||||
return;
|
||||
}
|
||||
|
||||
// System bytes match but stream/function disagree: protocol violation
|
||||
// (peer replied with the wrong SxFy). Surface this as a real diagnostic
|
||||
// rather than silently falling through to the primary handler.
|
||||
if (it != pending_requests_.end()) {
|
||||
log("!! unexpected " + h.describe() + " for pending sys=" +
|
||||
std::to_string(h.system_bytes) + " (expected S" +
|
||||
std::to_string(it->second.expected_stream) + "F" +
|
||||
std::to_string(it->second.expected_function) + ")");
|
||||
// Treat the data as a primary message anyway so the application
|
||||
// handler can deal with it; the pending request stays open until T3.
|
||||
}
|
||||
|
||||
// Primary message; only valid while SELECTED.
|
||||
if (state_ != State::Selected) {
|
||||
send_frame(Frame(Header::control(SType::RejectReq, h.system_bytes, kControlSessionId,
|
||||
@@ -267,7 +283,11 @@ void Connection::send_request(secs2::Message msg, ReplyHandler cb) {
|
||||
|
||||
auto t3 = std::make_shared<asio::steady_timer>(socket_.get_executor());
|
||||
t3->expires_after(timers_.t3);
|
||||
pending_requests_.emplace(sys, PendingRequest{std::move(cb), t3});
|
||||
// SECS-II reply convention: same stream, function+1.
|
||||
pending_requests_.emplace(sys, PendingRequest{
|
||||
/*expected_stream=*/msg.stream,
|
||||
/*expected_function=*/static_cast<uint8_t>(msg.function + 1),
|
||||
std::move(cb), t3});
|
||||
|
||||
auto self = shared_from_this();
|
||||
t3->async_wait([this, self, sys](std::error_code ec) {
|
||||
|
||||
Reference in New Issue
Block a user