(m)TLS to PG#627
Draft
shortishly wants to merge 173 commits into
Draft
Conversation
This change refactors the consumer protocol types (like MemberMetadata, MemberAssignment) from the broker crate into the sans-io crate. These are protocol-level types that belong with the other protocol serialization code. The sans-io crate now handles all encoding/decoding of the consumer protocol types independent of IO concerns.
The commit adds support for consumer groups including: - Client-side Group data structure - Range partition assignor implementation - Consumer protocol (de)serialization - Consumer group assignor trait and types
Bumps [actions/download-artifact](https://github.com/actions/download-artifact) from 7 to 8. - [Release notes](https://github.com/actions/download-artifact/releases) - [Commits](actions/download-artifact@v7...v8) --- updated-dependencies: - dependency-name: actions/download-artifact dependency-version: '8' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com>
Bumps [actions/upload-artifact](https://github.com/actions/upload-artifact) from 6 to 7. - [Release notes](https://github.com/actions/upload-artifact/releases) - [Commits](actions/upload-artifact@v6...v7) --- updated-dependencies: - dependency-name: actions/upload-artifact dependency-version: '7' dependency-type: direct:production update-type: version-update:semver-major ... Signed-off-by: dependabot[bot] <support@github.com>
Run the librdkafka v2.14.2 integration test suite against tansu as a client-compatibility check: just compat-librdkafka builds tansu, starts a memory:// broker advertising 127.0.0.1 (librdkafka resolves localhost to ::1), builds the librdkafka test-runner under target/compat/, and runs the 16 tests in compat/librdkafka/tests.allow. A compat-librdkafka CI job runs the same recipe. Only tests that create topics through the Admin API can run: tansu does not implement automatic topic creation on MetadataRequest, which the rest of the suite assumes. FINDINGS.md records the gaps found in the initial sweep, including CreatePartitions returning an empty topics array, ListOffsets queries hanging the client, OffsetCommit metadata not round-tripped, and InitProducerId not validating transaction.timeout.ms. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
MetadataService now creates topics that are named in a MetadataRequest but do not exist, when allow_auto_topic_creation permits it (versions prior to 4 do not have the field and behave as if it were true), then re-queries so the client receives full metadata in the same response. Topics are created with four partitions (the num.partitions the Kafka client test suites assume) and replication factor 1. Invalid topic names are reported as INVALID_TOPIC_EXCEPTION and not created. Also implement IncrementalAlterConfigs APPEND/SUBTRACT in dynostore with comma separated list semantics: the previous todo!() panicked while holding the topics lock, poisoning it so every subsequent request failed until restart. The same todo!() remains in the pg, lite and limbo backends. This grows the librdkafka compatibility suite from 16 to 74 passing tests, covering produce/consume, compression, consumer groups, idempotence, headers and admin paths. run.sh now waits for the broker to listen before running tests: with the librdkafka build cached, the first test could outrun broker startup. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…ment-payload-franz-go-fails-to-parse-with-response-did-not-contain-enough-data-to-be-valid
…member_assignment-payload-franz-go-fails-to-parse-with-response-did-not-contain-enough-data-to-be-valid join/sync: delay and retry request for members without an assignment
Run the franz-go v1.21.3 pkg/kgo integration test suite against tansu as a client-compatibility check, alongside the librdkafka suite: just compat-franz-go builds tansu, starts a memory:// broker, clones franz-go under target/compat/, and runs the 59 tests in compat/franz-go/tests.allow in a single go test invocation with KGO_TEST_RF=1 (tansu is single-node). A compat-franz-go CI job runs the same recipe. FINDINGS.md records the gaps behind the five excluded tests, most notably that Fetch omits the record batch containing a mid-batch offset instead of returning it whole: records before the next batch boundary are unreachable and a consumer positioned inside the last batch hangs forever. The others: UNKNOWN_MEMBER_ID under concurrent rebalances, aborted transactional records visible to read_committed consumers, and max.message.bytes not enforced on Produce. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
A fetch offset falling inside a batch returned nothing from that batch: the dynostore and slatedb engines selected batches by base offset alone, so the batch containing the fetch offset was omitted unless the offset was exactly its base. Kafka returns that batch whole and the client skips the records below the fetch offset. Records before the next batch boundary were unreachable, and a consumer positioned inside the last batch of a partition polled empty forever - hanging franz-go's TestAddRemovePartitions and TestIssue865, now in tests.allow (61 of 64 run). dynostore now also considers the greatest base offset preceding the fetch offset, dropping that batch after decoding if it ends before the fetch offset. slatedb has forward-only iteration, so the scan start is located with a binary search over least-key-at-or-after probes; fetches at or beyond the high watermark return early. The pg and lite engines rebuild batches from per-record rows and were not affected. The mid_batch regression test fetches at every offset of a two-batch partition against every engine, asserting all records from the fetch offset to the high watermark are returned and that only batches ending before the fetch offset are omitted. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
…ecord In lite/limbo fetch, the first record's header lookup used the caller's requested offset instead of that record's own offset_id. This matters once an earlier record at that offset has been removed (e.g. by cleanup.policy=compact), causing the surviving record to be returned with empty or mismatched headers. Also add tests covering kv/header round-tripping and a compaction scenario that exercises the fixed first-row path.
generation_id was incremented on every join, even rebalance pauses that returned the existing generation; now it starts at 0 and is only advanced by the coordinator's own logic, with Default::default() used in place of -1 sentinels in administrator.rs and lib.rs. sync_group now validates both that the syncing member is the leader and that assignments cover every current member, rather than only checking the leader, returning RebalanceInProgress otherwise. Also adjusts fetch byte-budget accounting in tansu-storage to track remaining max_bytes per partition once a non-empty batch has been returned, fixes a libsql panic on unexpected row counts to return an error instead, and simplifies the test storage_container/register_broker helpers to take borrowed/cloned args instead of generics. Adds tansu-proxy/src/bin/la.rs, a CLI for analyzing captured proxy logs. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
These are expected outcomes of idempotent producer retries, not failures, so logging them at error level was noisy. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…ersion Generate an IntoVersion impl for every message/body type that nulls out optional tagged fields whose version range excludes the request's api_version, and apply it in Frame::request/response before encoding. Without this, fields like FetchResponse::node_endpoints could be serialized on versions that don't support them, producing invalid wire output. Adds debug tracing for tag encoding and a regression test for FetchResponse v12 with node endpoints. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Parameterize the compat-franz-go just recipe with storage engine and feature flags so CI can exercise the franz-go compat suite against both memory:// and sqlite:// backends.
disbale franz until more of that is working properly
…termark
The watermark query selects three values:
select w.low, w.high, s.offset
For the read_committed boundary (last_stable), offset_stage picked w.high --
the end of the log -- instead of s.offset, the stable offset up to which
everything is committed. So read_committed consumers were handed the end of
the log and could read records from transactions that hadn't committed yet.
Read s.offset instead on all three backends (Postgres, libSQL, Turso), with a
regression test on each (the Turso one is skipped until that engine is mature
enough to run it).
fix(storage): stop read_committed from reading past the last stable offset
Move the transaction-timeout sweep out of Storage::maintain into a new Storage::maintain_transactions(now), run by the broker as its own task on its own interval. General maintenance defaults to 10 minutes -- too slow to release the Last Stable Offset an abandoned transaction pins. The sweep gets its own transaction_maintenance_interval (storage-URL query param), defaulting to 10s to match Kafka's transaction.abort.timed.out.transaction.cleanup.interval.ms. Both intervals skip missed ticks and guard against overlap (an AbortHandle per sweep), so a sweep slower than its interval can't run twice at once or pile up. StorageContainer::maintain_transactions records the same request/error metrics and span as the other dispatch methods. all backends for now return Ok().
refactor(storage): split the timeout sweep into maintain_transactions
`Authentication::is_authenticated()` returned `true` for any `Stage::Finished(_)`, ignoring whether validation produced an identity. rsasl's SCRAM server reaches `State::Finished` even on a bad client proof (it emits an `e=` error token and relies on a well-behaved client to abort), so a client that ran the SCRAM exchange with an invalid proof -- a wrong password, or a user with no stored credential -- and simply did not abort was admitted to gated APIs. Require `Stage::Finished(Some(_))`. Adds an enforcement test that drives that exact attack and asserts the connection is rejected with NotAuthenticated.
fix(auth): require a validated identity to treat SASL as authenticated
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
work in progress