diff --git a/src/zm_stream_socket.cpp b/src/zm_stream_socket.cpp index 63e199382..5d7cdd2e7 100644 --- a/src/zm_stream_socket.cpp +++ b/src/zm_stream_socket.cpp @@ -22,8 +22,10 @@ #include "zm_utils.h" #include +#include #include #include +#include #include #include #include @@ -133,9 +135,15 @@ std::vector StreamSocket::ParseAllowedUids(const std::string &value) { std::string trimmed = Trim(token, " \t"); if (trimmed.empty()) continue; + errno = 0; char *end = nullptr; unsigned long uid = strtoul(trimmed.c_str(), &end, 10); - if (end and *end == '\0') { + // strtoul wraps a leading '-' and saturates on overflow; reject both, and + // any value that does not round-trip through uid_t, so a typo cannot + // silently allow a different uid (e.g. 2^32 truncating to uid 0). + bool out_of_range = errno == ERANGE + or uid != static_cast(static_cast(uid)); + if (trimmed[0] != '-' and end != trimmed.c_str() and *end == '\0' and !out_of_range) { uids.push_back(static_cast(uid)); } else { Warning("StreamSocket: ignoring malformed uid '%s' in allowed uids", trimmed.c_str()); @@ -232,6 +240,29 @@ void StreamSocket::SetAudioParams(const AVCodecParameters *par) { Wake(); } +void StreamSocket::ClearAudioParams() { + { + std::lock_guard lock(mutex_); + if (hello_audio_payload_.empty()) + return; // no audio was announced; nothing to forget + hello_audio_payload_.clear(); + hello_audio_.reset(); + sequence_[static_cast(StreamId::Audio)] = 0; + // A dropped stream is a parameter change like any other: bump the + // generation so consumers re-init, and re-issue the surviving video HELLO + // under it. + ++generation_; + Info("StreamSocket: monitor %u audio stream removed, generation now %u", + monitor_id_, generation_); + if (!hello_video_payload_.empty()) { + hello_video_ = MakeMessage(MessageType::Hello, StreamId::Video, 0, 0, 0, + std::vector(hello_video_payload_), true); + BroadcastLocked(hello_video_); + } + } + Wake(); +} + void StreamSocket::SendMedia(const AVPacket *packet, StreamId stream, bool keyframe, int64_t pts_us) { if (!packet or packet->size <= 0) @@ -240,6 +271,16 @@ void StreamSocket::SendMedia(const AVPacket *packet, StreamId stream, bool video_keyframe = keyframe and stream == StreamId::Video; std::unique_lock lock(mutex_); + + // A stream is on the wire only once its parameters are announced. Without a + // HELLO a consumer cannot decode the payload (no codec id or extradata), so + // dropping here keeps the protocol's "HELLO precedes MEDIA" guarantee for + // e.g. audio packets a monitor forwards before record_audio announces them. + const std::vector &hello = + stream == StreamId::Audio ? hello_audio_payload_ : hello_video_payload_; + if (hello.empty()) + return; + uint32_t sequence = sequence_[static_cast(stream)]++; bool have_clients = !clients_.empty(); @@ -257,7 +298,7 @@ void StreamSocket::SendMedia(const AVPacket *packet, StreamId stream, header.flags = video_keyframe ? kFlagKeyframe : 0; header.sequence = sequence; header.generation = generation_; - header.pts_us = static_cast(pts_us); + header.pts_us = pts_us; // Header.pts_us is signed if (video_keyframe) { // Cache the keyframe for fast-start of late joiners; the cache references @@ -327,7 +368,7 @@ StreamSocket::MessagePtr StreamSocket::MakeMessage( header.flags = flags; header.sequence = sequence; header.generation = generation_; - header.pts_us = static_cast(pts_us); + header.pts_us = pts_us; // Header.pts_us is signed SerializeHeader(header, message->header.data()); message->blob_payload = std::move(payload); message->control = control; diff --git a/src/zm_stream_socket.h b/src/zm_stream_socket.h index 208bf46cd..470a797df 100644 --- a/src/zm_stream_socket.h +++ b/src/zm_stream_socket.h @@ -72,6 +72,12 @@ class StreamSocket { void SetVideoParams(const AVCodecParameters *par, AVRational frame_rate); void SetAudioParams(const AVCodecParameters *par); + // Forget the audio stream: drops the cached audio HELLO and stops announcing + // audio to new consumers. Called when a re-prime finds no audio stream (a + // camera that lost audio, or record_audio turned off) so a stale HELLO is not + // replayed and the generation bumps like any other parameter change. + void ClearAudioParams(); + // Queue one access unit / audio packet to every connected client. // pts_us must be in AV_TIME_BASE_Q. The packet's payload buffer is // reference-counted, not copied. Never blocks. diff --git a/src/zm_stream_socket_client.cpp b/src/zm_stream_socket_client.cpp index 23de5523f..35d2ac437 100644 --- a/src/zm_stream_socket_client.cpp +++ b/src/zm_stream_socket_client.cpp @@ -89,16 +89,18 @@ void StreamSocketClient::Run() { } ++consecutive_failures_; ::close(fd); - // sleep in small steps so Stop() stays responsive - auto deadline = std::chrono::steady_clock::now() + kReconnectDelay; - while (!terminate_ and std::chrono::steady_clock::now() < deadline) - std::this_thread::sleep_for(std::chrono::milliseconds(100)); + SleepBeforeRetry(); continue; } - Info("StreamSocketClient: connected to %s", path_.c_str()); - consecutive_failures_ = 0; + if (consecutive_failures_ == 0) { + Info("StreamSocketClient: connected to %s", path_.c_str()); + } else { + Debug(1, "StreamSocketClient: connected to %s (attempt %d)", + path_.c_str(), consecutive_failures_ + 1); + } setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &kReadTimeout, sizeof(kReadTimeout)); + session_messages_ = 0; connected_ = true; ReadLoop(fd); connected_ = false; @@ -106,12 +108,34 @@ void StreamSocketClient::Run() { if (terminate_) break; + + if (session_messages_ == 0) { + // Accepted and then closed before a single message: the producer + // rejected us (uid allow-list, client limit) or died mid-handshake. + // Back off like a failed connect, or a rejecting producer turns this + // loop into a connect storm. + if (consecutive_failures_ == 0 or consecutive_failures_ % 30 == 29) { + Warning("StreamSocketClient: %s closed the connection before any message" + " (attempt %d), retrying", path_.c_str(), consecutive_failures_ + 1); + } + ++consecutive_failures_; + SleepBeforeRetry(); + continue; + } + + consecutive_failures_ = 0; Debug(1, "StreamSocketClient: disconnected from %s, reconnecting", path_.c_str()); if (callbacks_.on_disconnect) callbacks_.on_disconnect(); } } +void StreamSocketClient::SleepBeforeRetry() { + auto deadline = std::chrono::steady_clock::now() + kReconnectDelay; + while (!terminate_ and std::chrono::steady_clock::now() < deadline) + std::this_thread::sleep_for(std::chrono::milliseconds(100)); +} + bool StreamSocketClient::ReadExact(int fd, uint8_t *out, size_t len) { size_t done = 0; while (done < len) { @@ -149,6 +173,7 @@ bool StreamSocketClient::ReadLoop(int fd) { if (payload_size > 0 and !ReadExact(fd, payload_.data(), payload_size)) return false; + ++session_messages_; Dispatch(header, payload_.data(), payload_size); } return true; diff --git a/src/zm_stream_socket_client.h b/src/zm_stream_socket_client.h index d2aa4be6b..400685597 100644 --- a/src/zm_stream_socket_client.h +++ b/src/zm_stream_socket_client.h @@ -51,7 +51,10 @@ class StreamSocketClient { std::function on_event; std::function on_bye; - std::function on_disconnect; // EOF or read error; reconnect follows + // EOF or read error after at least one message was received (an adopted + // descriptor reports every EOF). A connection the producer closes before + // any message - a rejection - retries with backoff and is not reported. + std::function on_disconnect; }; // Connect (with retry) to a stream socket path. @@ -73,11 +76,14 @@ class StreamSocketClient { bool ReadExact(int fd, uint8_t *out, size_t len); void Dispatch(const zm::stream_socket::Header &header, const uint8_t *payload, size_t size); + // Waits kReconnectDelay in small steps so Stop() stays responsive. + void SleepBeforeRetry(); std::string path_; Callbacks callbacks_; int adopted_fd_ = -1; int consecutive_failures_ = 0; + uint64_t session_messages_ = 0; // messages received on the current connection std::thread thread_; std::atomic terminate_{false}; std::atomic connected_{false}; diff --git a/src/zm_stream_socket_protocol.cpp b/src/zm_stream_socket_protocol.cpp index 98fd2b09f..7f6bbf05d 100644 --- a/src/zm_stream_socket_protocol.cpp +++ b/src/zm_stream_socket_protocol.cpp @@ -97,7 +97,7 @@ void SerializeHeader(const Header &header, uint8_t out[kHeaderSize]) { out[7] = header.flags; put_u32(out + 8, header.sequence); put_u32(out + 12, header.generation); - put_u64(out + 16, header.pts_us); + put_u64(out + 16, static_cast(header.pts_us)); // two's-complement } bool ParseHeader(const uint8_t in[kHeaderSize], Header &header) { @@ -108,7 +108,7 @@ bool ParseHeader(const uint8_t in[kHeaderSize], Header &header) { header.flags = in[7]; header.sequence = get_u32(in + 8); header.generation = get_u32(in + 12); - header.pts_us = get_u64(in + 16); + header.pts_us = static_cast(get_u64(in + 16)); // two's-complement if (header.version != kProtocolVersion) return false; diff --git a/src/zm_stream_socket_protocol.h b/src/zm_stream_socket_protocol.h index abad4e395..e84b3fb69 100644 --- a/src/zm_stream_socket_protocol.h +++ b/src/zm_stream_socket_protocol.h @@ -38,7 +38,8 @@ extern "C" { // u8 flags bit 0: keyframe (video); other bits reserved, must be 0 // u32 sequence per-stream, counts every message produced (drops appear as gaps) // u32 generation stream epoch; a bump means re-init the decoder from a new HELLO -// u64 pts_us microseconds, AV_TIME_BASE_Q, shared clock for both streams +// i64 pts_us signed microseconds (two's-complement), AV_TIME_BASE_Q, +// shared clock for both streams; AV_NOPTS_VALUE = unknown // [payload] // // HELLO payload is a TLV list (u8 tag, u16 length, value); unknown tags must @@ -97,7 +98,7 @@ struct Header { uint8_t flags; uint32_t sequence; uint32_t generation; - uint64_t pts_us; + int64_t pts_us; // signed; AV_NOPTS_VALUE means unknown uint32_t payload_size() const { return length - kHeaderLengthBytes; } }; diff --git a/tests/zm_stream_socket.cpp b/tests/zm_stream_socket.cpp index 3da434061..6b0481004 100644 --- a/tests/zm_stream_socket.cpp +++ b/tests/zm_stream_socket.cpp @@ -104,6 +104,14 @@ codec_parameters_ptr make_h264_parameters() { return par; } +codec_parameters_ptr make_aac_parameters() { + codec_parameters_ptr par{avcodec_parameters_alloc()}; + par->codec_type = AVMEDIA_TYPE_AUDIO; + par->codec_id = AV_CODEC_ID_AAC; + par->sample_rate = 48000; + return par; +} + } // namespace TEST_CASE("StreamSocket lifecycle", "[stream_socket]") { @@ -483,6 +491,94 @@ TEST_CASE("StreamSocket::ParseAllowedUids", "[stream_socket]") { REQUIRE(StreamSocket::ParseAllowedUids(" 33 , 1000 ") == std::vector{33, 1000}); REQUIRE(StreamSocket::ParseAllowedUids("33,,1000") == std::vector{33, 1000}); REQUIRE(StreamSocket::ParseAllowedUids("33,bogus,1000") == std::vector{33, 1000}); + + // Values that cannot be a uid are rejected, never wrapped or truncated + REQUIRE(StreamSocket::ParseAllowedUids("-1").empty()); + REQUIRE(StreamSocket::ParseAllowedUids("99999999999999999999").empty()); // > ULONG_MAX + if (sizeof(uid_t) < sizeof(unsigned long)) { + REQUIRE(StreamSocket::ParseAllowedUids("18446744073709551615").empty()); // ULONG_MAX + REQUIRE(StreamSocket::ParseAllowedUids("4294967296").empty()); // 2^32 would truncate to 0 + } + REQUIRE(StreamSocket::ParseAllowedUids("33,-1,1000") == std::vector{33, 1000}); +} + +TEST_CASE("StreamSocket drops media for a stream with no HELLO", "[stream_socket]") { + StreamSocket server(1, kSockPath); + REQUIRE(server.Start()); + + // Only video is announced; audio never gets a HELLO. + codec_parameters_ptr par = make_h264_parameters(); + server.SetVideoParams(par.get(), {0, 0}); + + TestClient client; + REQUIRE(client.Connect()); + ReceivedMessage hello; + REQUIRE(client.ReadMessage(hello)); + REQUIRE(hello.header.type == static_cast(MessageType::Hello)); + REQUIRE(hello.header.stream == static_cast(StreamId::Video)); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + + // An audio packet arrives before audio is announced: it must not reach the + // wire (a consumer could not decode it without a HELLO). + av_packet_ptr audio = make_packet(200, 0x33); + server.SendMedia(audio.get(), StreamId::Audio, false, 111); + // A following video packet does go out; if the audio had leaked we would + // read it first. + av_packet_ptr video = make_packet(200, 0x44); + server.SendMedia(video.get(), StreamId::Video, false, 222); + + ReceivedMessage message; + REQUIRE(client.ReadMessage(message)); + REQUIRE(message.header.type == static_cast(MessageType::Media)); + REQUIRE(message.header.stream == static_cast(StreamId::Video)); + REQUIRE(message.header.pts_us == 222); + + server.Stop(); +} + +TEST_CASE("StreamSocket::ClearAudioParams stops announcing audio", "[stream_socket]") { + StreamSocket server(1, kSockPath); + REQUIRE(server.Start()); + + codec_parameters_ptr vpar = make_h264_parameters(); + server.SetVideoParams(vpar.get(), {0, 0}); + codec_parameters_ptr apar = make_aac_parameters(); + server.SetAudioParams(apar.get()); + + // First consumer sees both HELLOs. + { + TestClient client; + REQUIRE(client.Connect()); + ReceivedMessage m; + REQUIRE(client.ReadMessage(m)); + REQUIRE(m.header.stream == static_cast(StreamId::Video)); + REQUIRE(client.ReadMessage(m)); + REQUIRE(m.header.stream == static_cast(StreamId::Audio)); + } + + // Audio goes away on a re-prime: generation bumps and the video HELLO is + // re-issued under it. + server.ClearAudioParams(); + + // A fresh consumer is told about video only, at the new generation. + TestClient client; + REQUIRE(client.Connect()); + ReceivedMessage video_hello; + REQUIRE(client.ReadMessage(video_hello)); + REQUIRE(video_hello.header.type == static_cast(MessageType::Hello)); + REQUIRE(video_hello.header.stream == static_cast(StreamId::Video)); + REQUIRE(video_hello.header.generation == 1); + + // No audio HELLO follows; a video media packet is the next thing on the wire. + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + av_packet_ptr video = make_packet(100, 0x66); + server.SendMedia(video.get(), StreamId::Video, false, 900); + ReceivedMessage next; + REQUIRE(client.ReadMessage(next)); + REQUIRE(next.header.type == static_cast(MessageType::Media)); + REQUIRE(next.header.stream == static_cast(StreamId::Video)); + + server.Stop(); } TEST_CASE("StreamSocket::InvalidateKeyframe stops replaying a stale keyframe", "[stream_socket]") { diff --git a/tests/zm_stream_socket_client.cpp b/tests/zm_stream_socket_client.cpp index 203c138f8..c0ecd7b70 100644 --- a/tests/zm_stream_socket_client.cpp +++ b/tests/zm_stream_socket_client.cpp @@ -28,7 +28,9 @@ #include #include #include +#include #include +#include #include #include #include @@ -273,6 +275,57 @@ TEST_CASE("StreamSocketClient skips unknown message types", "[stream_socket_clie client.Stop(); } +TEST_CASE("StreamSocketClient backs off when the producer rejects the connection", + "[stream_socket_client]") { + // A producer that accepts and immediately closes, as StreamSocket does for + // a client failing the uid allow-list or the client limit + ::unlink(kSockPath); + int listener = ::socket(AF_UNIX, SOCK_STREAM, 0); + REQUIRE(listener >= 0); + sockaddr_un addr = {}; + addr.sun_family = AF_UNIX; + strncpy(addr.sun_path, kSockPath, sizeof(addr.sun_path) - 1); + REQUIRE(::bind(listener, reinterpret_cast(&addr), sizeof(addr)) == 0); + REQUIRE(::listen(listener, 8) == 0); + + std::atomic accepts{0}; + std::atomic stop{false}; + std::thread rejecter([&] { + while (!stop) { + timeval tv = {0, 100000}; + fd_set fds; + FD_ZERO(&fds); + FD_SET(listener, &fds); + if (::select(listener + 1, &fds, nullptr, nullptr, &tv) > 0) { + int fd = ::accept(listener, nullptr, nullptr); + if (fd >= 0) { + ++accepts; + ::close(fd); + } + } + } + }); + + Collector collector; + { + StreamSocketClient client(std::string(kSockPath), collector.MakeCallbacks()); + std::this_thread::sleep_for(std::chrono::milliseconds(2500)); + client.Stop(); + } + stop = true; + rejecter.join(); + ::close(listener); + ::unlink(kSockPath); + + // With the 1s backoff after every rejected connection: the initial attempt + // plus two or three retries in 2.5s. Without backoff this is thousands. + REQUIRE(accepts >= 2); + REQUIRE(accepts <= 5); + // A rejection is not a session, so no disconnect was reported + std::lock_guard lock(collector.mutex); + REQUIRE(collector.disconnects == 0); +} + TEST_CASE("StreamSocketClient delivers EVENT frames", "[stream_socket_client]") { StreamSocket server(7, kSockPath); REQUIRE(server.Start()); diff --git a/tests/zm_stream_socket_protocol.cpp b/tests/zm_stream_socket_protocol.cpp index 8ff0755e8..a2aaea761 100644 --- a/tests/zm_stream_socket_protocol.cpp +++ b/tests/zm_stream_socket_protocol.cpp @@ -93,7 +93,7 @@ TEST_CASE("stream_socket::Header boundary values") { in.length = kHeaderLengthBytes; in.sequence = std::numeric_limits::max(); in.generation = std::numeric_limits::max(); - in.pts_us = std::numeric_limits::max(); + in.pts_us = std::numeric_limits::max(); SerializeHeader(in, wire); Header out = {}; REQUIRE(ParseHeader(wire, out)); @@ -102,6 +102,18 @@ TEST_CASE("stream_socket::Header boundary values") { REQUIRE(out.pts_us == in.pts_us); } + SECTION("negative and AV_NOPTS_VALUE pts survive the wire") { + in.length = kHeaderLengthBytes; + const int64_t values[] = {-1, -123456, std::numeric_limits::min()}; + for (int64_t pts : values) { + in.pts_us = pts; + SerializeHeader(in, wire); + Header out = {}; + REQUIRE(ParseHeader(wire, out)); + REQUIRE(out.pts_us == pts); + } + } + SECTION("length below fixed header is rejected") { in.length = kHeaderLengthBytes - 1; SerializeHeader(in, wire); diff --git a/tools/zm_stream_socket_dump.py b/tools/zm_stream_socket_dump.py index cb66a034e..a3a990f90 100755 --- a/tools/zm_stream_socket_dump.py +++ b/tools/zm_stream_socket_dump.py @@ -79,11 +79,12 @@ def main(): sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(20) sock.connect(path) - media_seen = {} + media_seen = {} # (stream, generation) -> last MEDIA sequence for _ in range(count): header = read_exact(sock, 24) + # pts is signed (two's-complement) and may be AV_NOPTS_VALUE length, version, mtype, stream, flags, seq, gen, pts = \ - struct.unpack('