SY-3556: Use high performance codecs for iterators

client/cpp/framer/framer.h

@@ -171,8 +171,6 @@ class StreamerConfig {
     std::vector<channel::Key> channels;
     /// @brief the downsample factor for the streamer.
     int downsample_factor = 1;
-    /// @brief enable experimental high-performance codec for the writer.
-    bool enable_experimental_codec = true;
     /// @brief writer group IDs whose frames should be filtered out by the server. Used
     /// for telemetry bypass deduplication.
     std::vector<std::uint32_t> exclude_groups;
@@ -239,8 +237,7 @@ class Streamer {
 
     StreamerConfig cfg;
 
-    /// @brief custom framing codec. only used when cfg.enable_experimental_codec is
-    /// set to true.
+    /// @brief custom framing codec.
     Codec codec;
 
     /// @brief throws if methods have been called on the streamer before it is open.
@@ -312,20 +309,6 @@ struct WriterConfig {
     /// Defaults to 1s when auto-commit is enabled.
     x::telem::TimeSpan auto_index_persist_interval = 1 * x::telem::SECOND;
 
-    /// @brief enable protobuf frame caching for the writer. This allows
-    /// the writer to avoid repeated allocation and deallocation of protobuf frames,
-    /// releasing significant heap pressure.
-    ///
-    /// @details IMPORTANT: This option should only be used for writers that write
-    /// a frame with the EXACT same dimensions on every write i.e. same number of
-    /// channels and series in the same order. Each series must have the same data
-    /// type and the same number of samples. BEHAVIOR IS UNDEFINED IF YOU DO NOT
-    /// FOLLOW THIS RULE.
-    bool enable_proto_frame_caching = false;
-
-    /// @brief enable experimental high-performance codec for the writer.
-    bool enable_experimental_codec = true;
-
 private:
     /// @brief binds the configuration fields to it's protobuf representation.
     [[nodiscard]] x::errors::Error to_proto(grpc::framer::WriterConfig *f) const;
@@ -416,8 +399,7 @@ class Writer {
     /// @brief the configuration used to open the writer.
     WriterConfig cfg;
 
-    /// @brief the custom synnax frame codec for encoding/decoding frames. This codec
-    /// is only used when cfg.enable_experimental_codec is true.
+    /// @brief the custom synnax frame codec for encoding/decoding frames.
     Codec codec;
     /// @brief the data buffer for storing encoded frames.
     std::vector<std::uint8_t> codec_data;
@@ -427,8 +409,6 @@ class Writer {
 
     /// @brief cached request for reuse during writes
     std::unique_ptr<grpc::framer::WriterRequest> cached_write_req;
-    /// @brief cached frame within the request for reuse
-    ::x::telem::pb::Frame *cached_frame = nullptr;
 
     /// @brief internal function that waits until an ack is received for a
     /// particular command.

client/cpp/framer/streamer.cpp

@@ -15,7 +15,6 @@ namespace synnax::framer {
 void StreamerConfig::to_proto(grpc::framer::StreamerRequest &f) const {
     f.mutable_keys()->Add(channels.begin(), channels.end());
     f.set_downsample_factor(downsample_factor);
-    f.set_enable_experimental_codec(enable_experimental_codec);
     f.mutable_exclude_groups()->Add(
         this->exclude_groups.begin(),
         this->exclude_groups.end()
@@ -31,11 +30,9 @@ Client::open_streamer(const StreamerConfig &config) const {
     if (!net_stream->send(req).ok()) net_stream->close_send();
     auto [_, res_err] = net_stream->receive();
     auto streamer = Streamer(std::move(net_stream), config);
-    if (config.enable_experimental_codec) {
-        streamer.codec = Codec(this->channel_client);
-        if (const auto codec_err = streamer.codec.update(config.channels))
-            return {Streamer(), codec_err};
-    }
+    streamer.codec = Codec(this->channel_client);
+    if (const auto codec_err = streamer.codec.update(config.channels))
+        return {Streamer(), codec_err};
     return {std::move(streamer), res_err};
 }
 

client/cpp/framer/writer.cpp

@@ -22,12 +22,8 @@ enum WriterCommand : uint32_t {
 };
 
 std::pair<Writer, x::errors::Error> Client::open_writer(const WriterConfig &cfg) const {
-    Codec codec;
-    if (cfg.enable_experimental_codec) {
-        codec = Codec(this->channel_client);
-        if (const auto codec_err = codec.update(cfg.channels))
-            return {Writer(), codec_err};
-    }
+    Codec codec(this->channel_client);
+    if (const auto codec_err = codec.update(cfg.channels)) return {Writer(), codec_err};
     auto [net_writer, err] = this->writer_client->stream("/frame/write");
     if (err) return {Writer(), err};
     grpc::framer::WriterRequest req;
@@ -99,30 +95,14 @@ x::errors::Error Writer::close() {
 }
 
 x::errors::Error Writer::init_request(const x::telem::Frame &fr) {
-    if (this->cfg.enable_experimental_codec) {
-        if (this->cached_write_req == nullptr)
-            this->cached_write_req = std::make_unique<grpc::framer::WriterRequest>();
-        this->cached_write_req->set_command(WRITE);
-        if (const auto err = this->codec.encode(fr, this->codec_data)) return err;
-        this->cached_write_req->set_buffer(
-            this->codec_data.data(),
-            this->codec_data.size()
-        );
-        return x::errors::NIL;
-    }
-
-    if (this->cached_write_req != nullptr && this->cfg.enable_proto_frame_caching) {
-        for (size_t i = 0; i < fr.series->size(); i++)
-            *cached_frame->mutable_series(
-                static_cast<int>(i)
-            ) = fr.series->at(i).to_proto();
-        return x::errors::NIL;
-    }
-    this->cached_write_req = nullptr;
-    this->cached_write_req = std::make_unique<grpc::framer::WriterRequest>();
+    if (this->cached_write_req == nullptr)
+        this->cached_write_req = std::make_unique<grpc::framer::WriterRequest>();
     this->cached_write_req->set_command(WRITE);
-    this->cached_frame = cached_write_req->mutable_frame();
-    *cached_frame = fr.to_proto();
+    if (const auto err = this->codec.encode(fr, this->codec_data)) return err;
+    this->cached_write_req->set_buffer(
+        this->codec_data.data(),
+        this->codec_data.size()
+    );
     return x::errors::NIL;
 }
 

client/py/synnax/framer/client.py

@@ -79,7 +79,6 @@ def open_writer(
         enable_auto_commit: bool = True,
         auto_index_persist_interval: TimeSpan = 1 * TimeSpan.SECOND,
         err_on_extra_chans: bool = True,
-        use_experimental_codec: bool = True,
     ) -> Writer:
         """Opens a new writer on the given channels.
 
@@ -128,7 +127,6 @@ def open_writer(
             err_on_unauthorized=err_on_unauthorized,
             enable_auto_commit=enable_auto_commit,
             auto_index_persist_interval=auto_index_persist_interval,
-            use_experimental_codec=use_experimental_codec,
         )
 
     def open_iterator(
@@ -302,7 +300,6 @@ def open_streamer(
         channels: channel.Params,
         downsample_factor: int = 1,
         throttle_rate: float = 0,
-        use_experimental_codec: bool = True,
         exclude_groups: list[int] | None = None,
     ) -> Streamer:
         """Opens a new streamer on the given channels. The streamer will immediately
@@ -323,7 +320,6 @@ def open_streamer(
             client=self.__stream_client,
             downsample_factor=downsample_factor,
             throttle_rate=throttle_rate,
-            use_experimental_codec=use_experimental_codec,
             exclude_groups=exclude_groups,
         )
 

client/py/synnax/framer/codec.py

@@ -268,24 +268,24 @@ def decode(self, data: bytes, offset: int = 0) -> FramePayload:
         keys = list()
         series_list = list()
 
-        for key in state.keys:
-            if not flags.all_channels_present:
-                if idx >= len(buffer):
-                    break
-                frame_key = struct.unpack_from("<I", buffer, idx)[0]
-                if frame_key != key:
-                    continue
-                idx += KEY_SIZE
-            data_type = state.data_types[key]
+        def decode_series(key: channel.Key) -> bool:
+            nonlocal idx
+            data_type = state.data_types.get(key)
+            if data_type is None:
+                return False
             curr_len = data_len
             if not flags.eq_len:
+                if idx + DATA_LENGTH_SIZE > len(buffer):
+                    return False
                 curr_len = struct.unpack_from("<I", buffer, idx)[0]
                 idx += DATA_LENGTH_SIZE
 
             data_byte_len = curr_len
             if not data_type.is_variable:
                 data_byte_len = curr_len * data_type.density
 
+            if idx + data_byte_len > len(buffer):
+                return False
             series_data = bytes(buffer[idx : idx + data_byte_len])
             idx += data_byte_len
 
@@ -294,12 +294,16 @@ def decode(self, data: bytes, offset: int = 0) -> FramePayload:
             elif flags.eq_tr:
                 tr = TimeRange(start=start_time, end=end_time)
             else:
+                if idx + TIME_RANGE_SIZE > len(buffer):
+                    return False
                 s, e = struct.unpack_from("<QQ", buffer, idx)
                 tr = TimeRange(start=s, end=e)
                 idx += TIME_RANGE_SIZE
 
             curr_alignment = alignment
             if not flags.eq_align and not flags.zero_alignments:
+                if idx + ALIGNMENT_SIZE > len(buffer):
+                    return False
                 curr_alignment = Alignment(struct.unpack_from("<Q", buffer, idx)[0])
                 idx += ALIGNMENT_SIZE
 
@@ -312,6 +316,18 @@ def decode(self, data: bytes, offset: int = 0) -> FramePayload:
                     alignment=curr_alignment,
                 )
             )
+            return True
+
+        if flags.all_channels_present:
+            for key in state.keys:
+                if not decode_series(key):
+                    break
+        else:
+            while idx + KEY_SIZE <= len(buffer):
+                frame_key = struct.unpack_from("<I", buffer, idx)[0]
+                idx += KEY_SIZE
+                if not decode_series(frame_key):
+                    break
 
         return FramePayload(keys=keys, series=series_list)
 

client/py/synnax/framer/iterator.py

@@ -10,14 +10,18 @@
 from __future__ import annotations
 
 from enum import Enum
+from typing import cast
 
 from pydantic import BaseModel
 
 import synnax.channel.payload as channel
 from alamos import NOOP, Instrumentation
-from freighter import EOF, Stream, StreamClient
+from freighter import EOF, ExceptionPayload, Stream, WebsocketClient
+from freighter.transport import P
+from freighter.websocket import Message
 from synnax.exceptions import UnexpectedError
 from synnax.framer.adapter import ReadFrameAdapter
+from synnax.framer.codec import LOW_PERF_SPECIAL_CHAR, WSFramerCodec
 from synnax.framer.frame import Frame, FramePayload
 from synnax.telem import TimeRange, TimeSpan, TimeStamp
 
@@ -56,10 +60,33 @@ class _Response(BaseModel):
     variant: _ResponseVariant
     command: _Command
     ack: bool
-    error: str | None
+    error: ExceptionPayload | None
     frame: FramePayload
 
 
+class WSIteratorCodec(WSFramerCodec):
+    def encode(self, data: BaseModel) -> bytes:
+        return self.lower_perf_codec.encode(data)
+
+    def decode(self, data: bytes, pld_t: type[P]) -> P:
+        if data[0] == LOW_PERF_SPECIAL_CHAR:
+            return self.lower_perf_codec.decode(data[1:], pld_t)
+        frame = self.codec.decode(data, 1)
+        return cast(
+            P,
+            Message(
+                type="data",
+                payload=_Response(
+                    variant=_ResponseVariant.DATA,
+                    command=_Command.OPEN,
+                    ack=False,
+                    error=None,
+                    frame=frame,
+                ),
+            ),
+        )
+
+
 class Iterator:
     """Used to iterate over a databases telemetry in time-order. It should not be
     instantiated directly, and should instead be instantiated using the segment Client.
@@ -82,7 +109,7 @@ class Iterator:
     def __init__(
         self,
         tr: TimeRange,
-        client: StreamClient,
+        client: WebsocketClient,
         adapter: ReadFrameAdapter,
         chunk_size: int = 100000,
         downsample_factor: int = 1,
@@ -91,6 +118,7 @@ def __init__(
         self.tr = tr
         self.instrumentation = instrumentation
         self.__adapter = adapter
+        client = client.with_codec(WSIteratorCodec(self.__adapter.codec))
         self.__stream = client.stream("/frame/iterate", _Request, _Response)
         self._chunk_size = chunk_size
         self._downsample_factor = downsample_factor
@@ -193,16 +221,20 @@ def close(self) -> None:
         exc = self.__stream.close_send()
         if exc is not None:
             raise exc
-        r, exc = self.__stream.receive()
-        if exc is None:
-            raise UnexpectedError(
-                f"""Unexpected missing close acknowledgement from server.
-                Please report this issue to the Synnax team.
-                Response: {r}
-                """
-            )
-        elif not isinstance(exc, EOF):
-            raise exc
+        while True:
+            r, exc = self.__stream.receive()
+            if r is not None:
+                continue
+            if exc is None:
+                raise UnexpectedError(
+                    f"""Unexpected missing close acknowledgement from server.
+                    Please report this issue to the Synnax team.
+                    Response: {r}
+                    """
+                )
+            if not isinstance(exc, EOF):
+                raise exc
+            break
 
     def __iter__(self) -> Iterator:
         self.seek_first()

client/py/synnax/framer/streamer.py

@@ -83,13 +83,10 @@ def __init__(
         adapter: ReadFrameAdapter,
         downsample_factor: int = 1,
         throttle_rate: float = 0,
-        use_experimental_codec: bool = True,
         exclude_groups: list[int] | None = None,
     ) -> None:
         self._adapter = adapter
-        if use_experimental_codec:
-            client = client.with_codec(WSStreamerCodec(self._adapter.codec))
-
+        client = client.with_codec(WSStreamerCodec(self._adapter.codec))
         self._stream = client.stream(_ENDPOINT, _Request, _Response)
         self._downsample_factor = downsample_factor
         self._throttle_rate = throttle_rate

client/py/synnax/framer/writer.py

@@ -189,13 +189,11 @@ def __init__(
         err_on_unauthorized: bool = False,
         enable_auto_commit: bool = True,
         auto_index_persist_interval: TimeSpan = 1 * TimeSpan.SECOND,
-        use_experimental_codec: bool = True,
         group: int = 0,
     ) -> None:
         self.start = start
         self._adapter = adapter
-        if use_experimental_codec:
-            client = client.with_codec(WSWriterCodec(adapter.codec))
+        client = client.with_codec(WSWriterCodec(adapter.codec))
         self._stream = client.stream("/frame/write", WriterRequest, WriterResponse)
         config = WriterConfig(
             control_subject=Subject(name=name, key=str(uuid4()), group=group),