Skip to content

Refactor[OBE-12143] - Build stateful decoders per unit from a DecoderFactory - #182

Open
janmejay-s1 wants to merge 2 commits into
masterfrom
OBE-12143__make_deserializer_stateful
Open

janmejay-s1 wants to merge 2 commits into
masterfrom
OBE-12143__make_deserializer_stateful

Conversation

@janmejay-s1

Copy link
Copy Markdown
Contributor

What

Decoders are now built per unit from a factory instead of being cloned.

  • Codecs API
    • DecodingConfig::build() returns a cloneable DecoderFactory<Stream | Datagram>.
    • Sources call build() on it once per unit (connection, datagram, message, request, S3/GCS object) to get a fresh Decoder.
    • format::Deserializer::parse takes &mut self.
    • The Framer/Deserializer traits drop DynClone and Sync.
    • Decoder, Framer and Deserializer are no longer Clone. static_assertions keeps them Send + !Clone + !Sync, with the reasoning written next to the assertions.
  • Factories
    • FramerFactory and DeserializerFactory mirror the existing enums.
    • Expensive preparation happens once per source, not per unit: VRL compile, protobuf descriptor load, avro schema parse and the json_paths rule trie.
    • State meant to be shared across units lives in the factory: netflow's template cache and chunked GELF's partial-message map.
    • DecoderFactory<Datagram>::build(peer) scopes netflow templates per exporter; Stream decoders get a fresh scope.
    • DecoderFactory<S> is bounded by a sealed Scope trait.
  • json_paths: owns its streaming parser state directly. No Arc<Mutex>, and no hand-written state-resetting Clone.
  • VRL: the deserializer reuses one Runtime per decoder and clears it after each parse.
  • Migrated sources:
    • socket tcp/udp/unix, syslog unix, statsd unix (build_unix_stream_source now takes a per-connection builder)
    • http_server, heroku_logs, datadog_agent, aws_kinesis_firehose, http_client
    • kafka, nats, pulsar, redis, websocket, gcp_pubsub, aws_sqs, amqp
    • exec, demo_logs, file_descriptors, aws_s3, the component-validation HTTP resource
  • Behaviour fixes and changes
    • aws_s3: now builds one decoder per object. Previously one deserializer was shared across every object and across concurrent SQS tasks, so a stateful codec such as json_paths could carry a half-read document from one object into the next.
    • amqp: builds its factory once, before connecting, instead of rebuilding the whole decoding config (including VRL compilation) on every delivery. An invalid decoding config now fails source build.
    • Private submodule: bumped for the matching Observo source changes.

Why

Stateful deserializers needed interior mutability and a hand-written Clone that reset their state. Whether a clone copied, reset or shared state was decided implicitly by each codec's Clone impl, and two sources got it wrong. Making decoders built per unit, and impossible to clone, puts that decision in one place: the factory.

How to Test

  1. TMPDIR=/tmp cargo nextest run -p codecs (including --features syslog,observo).
  2. Run the per-unit isolation tests, which pass on the old code and fail if decoder state is shared:
    cargo nextest run --no-default-features --features sources-heroku_logs,sources-datadog_agent,sources-aws_kinesis_firehose,sources-http_client,sources-socket,sources-exec,transforms-route --lib -- do_not_share_json_paths_state
  3. Run the factory sharing tests in codecs: factory, chunked_gelf_*_decoders_* and netflow_*_decoders_*.
  4. FEATURES=observo,observo-test,transforms-route make test. transforms-route is needed to avoid an existing is_indexable dead-code error that this branch doesn't touch.

Verification notes

  • make test: passes, except for compliance tests with fixed 1–5 s windows that fail under full-run load and pass on their own:
    • http_client and prometheus scrape
    • one Observo source test
    • journald excludes_matches
  • prometheus pushgateway happy-path tests: fail intermittently in one checkout only. The same commit passes 12/12 from a clean worktree, as does the base commit.
  • One Observo test: fails because its config uses a console sink that isn't compiled into the Observo feature set.
  • Clippy: make check-clippy reports existing errors under the 1.88 toolchain; the new lines add no findings.

Breaking Changes

Public codecs API changes, for code outside this repo that uses the decoding API:

  • DecodingConfig::build(), DeserializerConfig::build() and FramingConfig::build(limits) return factories. The configs for VRL, protobuf, avro, json_paths, netflow and chunked GELF also return factories from build().
  • Decoder, Framer, Deserializer, NetflowDecoder and ChunkedGelfDecoder are no longer Clone.
  • Decoder::clone_for_connection, Decoder::clone_for_datagram_peer, NetflowDecoder::set_datagram_peer and NetflowDecoder::set_new_connection_scope are removed. Use DecoderFactory::build.
  • format::Deserializer::parse takes &mut self.
  • build_unix_stream_source takes impl Fn() -> Decoder instead of a Decoder.

Jira: OBE-12143

🤖 Generated with Claude Code

janmejay-s1 and others added 2 commits November 6, 2026 11:34
…Factory

DecodingConfig::build() now returns a cloneable DecoderFactory<Stream |
Datagram>, and sources build a fresh Decoder from it per unit (connection,
datagram, message, request, object). Deserializer::parse takes &mut self, the
Framer and Deserializer traits drop DynClone and Sync, and Decoder, Framer and
Deserializer are no longer Clone; static assertions keep them Send, !Clone and
!Sync.

Expensive preparation (VRL compile, protobuf descriptor load, avro schema
parse, json_paths rules) happens once in the factory, and state shared across
units (netflow's template cache, chunked GELF's partial-message map) lives
there too. json_paths owns its parser state directly, and the VRL deserializer
reuses one runtime per decoder.

aws_s3 now builds a decoder per object instead of sharing one deserializer
across objects and concurrent SQS tasks, and amqp builds its factory once
instead of rebuilding the decoding config per delivery. Bumps the private
submodule for the matching Observo source changes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- The HTTP client loop no longer clones its context builder per request:
  build() takes &self, so neither the clone nor the Clone bounds on call()
  and HttpClientContextBuilder are needed.
- Netflow, chunked GELF and protobuf factories are built directly instead of
  constructing a throwaway decoder (which, for netflow, also consumed a
  connection id) and taking its parts; the instance constructors now delegate
  to the factories.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
@janmejay-s1
janmejay-s1 requested a review from jsbalis1 November 6, 2026 07:40
NetflowDecoder::new_with_max_length(max_length)
} else {
NetflowDecoder::new()
/// Build a `NetflowDecoderFactory` from this configuration.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Netflow seems like a misfit in framing. The way it works is it receives packet of types data as well as templates And in a data packet you ref those templates to get final values.
We should actually move this to a proper source and move it from framing.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants