[novatechflow]

Fixes #318

The LFS proxy HTTP API (/lfs/produce and /lfs/uploads) only used the static spec.lfsProxy.backends list (KAFSCALE_PROXY_BACKENDS) to find brokers. Without it, every produce returned 503 "no backends available". It also uploaded the object to S3 first, so each failed produce left an orphan object.

Changes:

• The LFS module now uses the brokers registered in etcd when no backends are configured. This is the same discovery the Kafka listener of the proxy already uses (currentBackends). Explicit backends still take priority.
• When etcd lists no brokers, there is no broker. Only when etcd cannot be read does the module use the brokers it last found, and only while the backend cache TTL (spec.lfsProxy.backendCacheTTLSeconds, default 60 seconds) has not passed, as the Kafka listener does.
• A failed etcd read no longer counts as a fresh broker list. syncStoreSnapshot only logged the error, so currentBackends returned the cached brokers and refreshed lastHealthy, and the backend cache TTL never applied during an etcd outage. This was already the case on main for the Kafka listener and the readiness check. The error is now returned, lastHealthy only moves on a successful read, and all callers use the cached brokers only while the cache is fresh.
• The Kafka listener applies the same TTL. Past it without a successful etcd read, the listener answers no Metadata or FindCoordinator request from the cached snapshot and routes no produce, fetch, group or other request, neither over cached partition or group routes nor over broker connections that are already open. Before this, cached routes, broker addresses, pooled connections and the in-memory snapshot kept serving with no limit. The check is one function, listenerRoutable, used in the request loop next to the existing readiness check, with a backstop in connectForAddr and handleMetadata. Each request gets the existing not-ready response: REQUEST_TIMED_OUT on the response or on each topic, partition, group or resource it names. A Metadata request for all topics names none, so it gets an empty reply with no brokers, no topics and no error code, which clients treat as failed metadata and retry. A request type with no not-ready response gets no answer. The client connection is then closed. ApiVersions is still answered. A stale cache gets one etcd read before a request is rejected, so the listener serves again as soon as etcd answers. Static backends are not affected.
• /lfs/produce and /lfs/uploads connect to a broker before they upload anything or start a multipart upload. If none can be reached, for example when etcd still lists brokers that were scaled down, they return 503 backend_unavailable and touch nothing in S3. The produce uses the connection from that check, so a produce opens one broker connection. A multipart upload keeps it in a small idle set (up to 4 connections, at most 1 minute old) and reuses it at completion when it is still open on the same proxy.
• /lfs/produce checks the TTL again after the upload, right before the produce, when brokers come from etcd. If it has passed, the object is deleted and the request gets 503 backend_unavailable. Before, only the socket was checked after the upload, so a long upload that outlasted the TTL still sent the record. With etcd down, the existing topic generation check after the upload already fails first with 503 metadata_unavailable and deletes the object, so this is a second guard. The multipart completion connects right before its produce, so it already checks the TTL there.
• If a broker goes away after the object was uploaded, the proxy sees the closed connection before the produce and connects again. If it cannot connect to any broker, it deletes the object. If that delete fails, the object is counted as an orphan as before.
• With brokers from etcd, /lfs/produce and multipart completion send the produce to the broker that owns the partition, from the same partition routes the Kafka listener uses (partitionOwnerAddr), and reuse an idle broker connection only when it goes to that broker. Before, they picked a broker round robin and reused any idle connection, so with two or more brokers a non-owner answered NOT_LEADER_OR_FOLLOWER, the request got 502 and the object was left as an orphan. When a broker answers NOT_LEADER_OR_FOLLOWER, LEADER_NOT_AVAILABLE, FENCED_LEADER_EPOCH or UNKNOWN_LEADER_EPOCH, the route is dropped and the produce goes to the current owner, to at most 3 brokers, with the retry wait ending on cancel and the TTL checked before each new attempt. When no broker accepts it, the object is deleted and the request gets 503 backend_unavailable. When the owner is not known yet, a listed broker is picked as before. Static backends pick a broker as before, and on that answer try another listed broker if there is one. With only one address, for example a broker Service, there is no other broker to try, so the request gets 503 and the object is deleted.
• When the produce request was already sent and then fails for another reason, the object is kept and counted as an orphan, as before. The broker may have written the record.
• The wait between broker retries ends when the request is cancelled and returns the context error, and there is no wait after the last attempt.
• Docs: docs/operations.md, docs/upgrade-compatibility.md and docs/releases/unreleased.md.

Behavior changes: a multipart upload without a reachable broker now fails at /lfs/uploads instead of at completion. A produce that no broker accepts as partition leader now gets 503 backend_unavailable and the object is deleted, instead of 502 with an orphan. When etcd cannot be read for longer than KAFSCALE_PROXY_BACKEND_CACHE_TTL_SEC (default 60 seconds), the proxy now reports NotReady, its Kafka listener answers no Metadata and routes no request until etcd answers again, and the LFS HTTP API sends nothing to a broker; on main it kept serving the last broker list, routes and metadata without a limit. The LFS HTTP API needs etcd for every request, because it reads the topic generation each time, so it returns 503 metadata_unavailable as soon as etcd cannot be read, even within the TTL. The TTL does not keep produces working either, because brokers stop acking produces while etcd is down. This is on purpose: when etcd is down, the cluster has to stop to avoid data loss, so past the TTL the proxy stops instead of letting clients write through stale routing. docs/upgrade-compatibility.md notes it too. Configurations with explicit backends also change on non-owner answers: NOT_LEADER_OR_FOLLOWER, LEADER_NOT_AVAILABLE, FENCED_LEADER_EPOCH or UNKNOWN_LEADER_EPOCH now give 503 backend_unavailable with the object deleted, where before they gave 502 and left an orphan. Otherwise explicit backends behave as before, and the TTL does not apply to them.

Tests (cmd/proxy/lfs_http_backends_test.go):

• TestLFSHTTPProduceUsesDiscoveredBrokersWithoutBackends
• TestLFSHTTPProduceExplicitBackendsWinOverDiscovery
• TestLFSHTTPProduceWithoutBackendSkipsUpload
• TestLFSHTTPProduceDeletesUploadWhenBackendDialFails
• TestLFSHTTPUploadInitWithoutBackendStartsNoUpload
• TestLFSHTTPMultipartDeletesUploadWhenBackendDialFails
• TestLFSHTTPProduceStaticBackendStillWorks
• TestLFSHTTPSkipsUploadAfterDiscoveredBrokersAreGone
• TestLFSHTTPDiscoveryErrorWithStaleCacheIsNoBackend
• TestLFSHTTPDiscoveryErrorWithFreshCacheStillWorks
• TestLFSHTTPSkipsUploadWhenListedBrokerIsUnreachable

Tests with an embedded etcd whose client is closed (cmd/proxy/lfs_etcd_outage_test.go):

• TestEtcdOutageWithStaleCacheIsNoBackend
• TestEtcdBlipWithFreshCacheStillServes

Tests for partition routing with two broker stubs, one owner and one non-owner, behind an embedded etcd with partition routes (cmd/proxy/lfs_partition_routing_test.go), each for /lfs/produce and multipart completion:

• TestLFSHTTPProduceRoutesToPartitionOwner
• TestLFSHTTPMultipartCompleteRoutesToPartitionOwner

Cases: owner on the first try; a stale route to the non-owner, then the owner; the owner moving during the request (for multipart, the idle connection from init to the old owner is not reused); no broker leading the partition (503, object deleted). All eight fail on a501f57 and pass with the fix.

Tests for the TTL after an upload (cmd/proxy/lfs_upload_ttl_test.go):

• TestLFSHTTPProduceChecksTTLAgainAfterUpload (fails on c368856, passes with the fix)
• TestLFSHTTPProduceStaticBackendsIgnoreTTLAfterUpload
• TestLFSHTTPMultipartCompleteChecksTTL

Tests for broker connections (cmd/proxy/lfs_backend_dials_test.go):

• TestLFSHTTPProduceDialsBrokerOnce
• TestLFSHTTPMultipartDialsBrokerOnce
• TestLFSConnectBackendReturnsOnCancelDuringBackoff

Tests for the Kafka listener with an embedded etcd, real partition and group routes, and a closed etcd client (cmd/proxy/listener_cache_ttl_test.go):

• TestListenerStopsRoutingPastCacheTTL (Metadata, FindCoordinator, Produce, Fetch, JoinGroup, ListOffsets, each sent once with etcd up and again on the same connection past the TTL)
• TestListenerServesDuringEtcdBlipWithinCacheTTL
• TestListenerRecoversWhenEtcdIsReadableAgain
• TestRoutingHelpersRefuseStaleCache

The new tests fail on main and pass with the fix, except TestListenerServesDuringEtcdBlipWithinCacheTTL, which passes on both and guards the behavior within the TTL. TestHandleHTTPUploadInit_Success now sets a backend that accepts connections. countingBackend in the generation tests counts connections that send a request, because the produce now connects before the upload.

Checked on kind with no spec.lfsProxy.backends set:

• An LFS HTTP produce returned 200.
• kcat with librdkafka 2.11.1 consumed the envelope record from the broker service.
• /lfs/download returned the payload with its integrity check.
• With the operator and brokers scaled to 0 and the proxy pod NotReady, requests sent straight to the pod with a port forward got 503 from two produces and two /lfs/uploads calls, with no new object in the bucket and no multipart upload in list-multipart-uploads.

Checked on kind for the Kafka listener with etcd scaled to 0 and backendCacheTTLSeconds: 20, sending kcat (librdkafka 2.11.1) straight to the proxy pod IP:

• Before the outage, kcat -L and a produce worked.
• 35 seconds after etcd went down, the pod was NotReady, kcat -L failed to get metadata and the produce failed.
• After etcd came back, the pod was Ready again within about 20 seconds and kcat -L worked. The record produced after recovery was stored (offset 1), but its ack took longer than the 15 second kcat delivery timeout.

[novatechflow]

QA of head 6f7340b against main bf2136c. One finding should be fixed before merge; everything else checks out.

Finding: the "known broker" check passes on a stale cache, so /lfs/uploads still starts an upload when no broker is running.
resolveBackends (cmd/proxy/lfs.go) falls back to m.cachedBackends whenever discovery returns an error or an empty list, and that cache never expires. The Kafka listener's fallback in connectBackend (cmd/proxy/main.go) only uses its cache while p.cacheFresh() holds. So once a proxy has seen a broker, waitForBackends keeps passing after every broker is gone, even when etcd plainly lists none.

On kind (2 proxy replicas, no spec.lfsProxy.backends) with the operator and the broker StatefulSet scaled to 0, waiting 70s until the proxy pods were NotReady with fetch probe: no backend, then calling a proxy pod directly:

• POST /lfs/produce returned 503 after about 3s with dial tcp: lookup kafscale-broker-0... no such host. The proxy did upload the object and then deleted it, so the bucket had no new object. That part works, but it is the delete path, not the skip the PR describes.
• POST /lfs/uploads returned 200 and started an S3 multipart upload. list-multipart-uploads showed it still in flight afterwards.

That contradicts the PR body and docs/releases/unreleased.md ("returns 503 backend_unavailable without uploading when there is none", "a multipart upload without a known broker now fails at the start") for the normal case where brokers go away after the proxy started. Suggested direction: when discovery succeeds with zero brokers, treat that as no backend; use the cache only when discovery itself fails, and only while it is fresh, matching the Kafka listener's TTL. Please add a red/green test where discovery first returns a broker and then returns an empty list, checking that produce returns 503 with no S3 put and upload init returns 503 with no StartMultipartUpload.

Checked and fine

• Red/green: with main's wiring (a test-only attachLFSModule shim on main), 5 of the 7 new tests fail; all pass on the head. go vet, gofmt and go test -race ./cmd/proxy/ pass; go build ./... passes.
• Wiring: attachLFSModule runs before the HTTP server starts. KAFSCALE_LFS_PROXY_BACKENDS and KAFSCALE_PROXY_BACKENDS keep the same precedence as on main, and the tracker still only uses the static list, as the operations.md line says. No exported API changes (package main).
• Kind, no spec.lfsProxy.backends: LFS HTTP produce 200, the envelope consumed with kcat (librdkafka 2.11.1) matches key and sha256, /lfs/download returns the payload with the integrity check. After scaling the broker back to 1, produce works again without a proxy restart.
• Kind, spec.lfsProxy.backends set to the broker service: both env vars are set, the tracker uses the static broker, and produce returns 200 through the Service and on a pod directly.
• handleHTTPUploadComplete completes the S3 upload before connectBackend, so deleting the object and the session on a failed dial is correct.

Unrelated, filed separately as #321: on kind, multipart part and complete calls through the Service hit the other proxy replica and got 404 upload_not_found in 3 of 4 tries, because upload sessions are kept in pod memory. Same on main, not caused by this PR.

[PaxMachinaOne]

🤖 PaxMachina automated review (worker-code)

The fallback can use brokers that discovery has authoritatively removed.

• Correctness: cmd/proxy/lfs.go:499 falls back to cachedBackends even when discovery succeeds but returns an empty broker list. After brokers are removed from etcd, the stale cache therefore passes the pre-upload check and may route production to a removed/reassigned broker; otherwise it uploads before the stale dial fails, contrary to the promised “upload nothing without a known broker” behavior. Cache fallback should apply only when discovery errors; a successful empty result should clear/ignore the cache and return errLFSNoBackends. Add a regression test that first discovers a broker, then updates metadata to an empty broker list.

[PaxMachinaOne]

The findings are fixed and verified

• Full proxy test suite passes.
• Routing tests pass under the race detector.
• Previous two-broker reproduction now passes.