Receive: fix capnproto replication in endless loop

Signed-off-by: Michał Mazur <[email protected]>

Author: lachruzam

CHANGELOG.md

@@ -10,6 +10,10 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 
 ## Unreleased
 
+### Fixed
+
+- [#8254](https://github.com/thanos-io/thanos/issues/8254) Receive: Endless loop of retried replication with capnproto and distributors
+
 ## [v0.40.0](https://github.com/thanos-io/thanos/tree/release-0.40) - 2025 10 20 (in progress)
 
 ### Fixed
@@ -22,22 +26,21 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 
 - [#8366](https://github.com/thanos-io/thanos/pull/8366) Store: optionally ignore Parquet migrated blocks
 - [#8359](https://github.com/thanos-io/thanos/pull/8359) Tools: add `--shipper.upload-compacted` flag for uploading compacted blocks to bucket upload-blocks
-- [#8484](https://github.com/thanos-io/thanos/pull/8484) Query: add `/api/v1/status/tsdb` API endpoint.
 
 ### Changed
 
 - [#8370](https://github.com/thanos-io/thanos/pull/8370) Query: announced labelset now reflects relabel-config
 
 ### Removed
 
-## [v0.39.2](https://github.com/thanos-io/thanos/tree/release-0.39) - 2025 07 17
+### [v0.39.2](https://github.com/thanos-io/thanos/tree/release-0.39) - 2025 07 17
 
 ### Fixed
 
 - [#8374](https://github.com/thanos-io/thanos/pull/8374) Query: fix panic when concurrently accessing annotations map
 - [#8375](https://github.com/thanos-io/thanos/pull/8375) Query: fix native histogram buckets in distributed queries
 
-## [v0.39.1](https://github.com/thanos-io/thanos/tree/release-0.39) - 2025 07 01
+### [v0.39.1](https://github.com/thanos-io/thanos/tree/release-0.39) - 2025 07 01
 
 Fixes a memory leak issue on query-frontend. The bug only affects v0.39.0.
 

pkg/receive/writecapnp/client.go

@@ -109,6 +109,15 @@ func (r *RemoteWriteClient) connect(ctx context.Context) error {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	if r.conn != nil {
+		s := capnp.Client(r.writer).State()
+		if !s.IsPromise {
+			if err, ok := s.Brand.Value.(error); ok {
+				// If bootstrap failed, close the connection to force a reconnect.
+				r.closeUnlocked()
+				return errors.Wrap(err, "client in error state")
+			}
+		}
+
 		return nil
 	}
 
@@ -123,11 +132,15 @@ func (r *RemoteWriteClient) connect(ctx context.Context) error {
 
 func (r *RemoteWriteClient) Close() error {
 	r.mu.Lock()
+	r.closeUnlocked()
+	r.mu.Unlock()
+	return nil
+}
+
+func (r *RemoteWriteClient) closeUnlocked() {
 	if r.conn != nil {
 		conn := r.conn
 		r.conn = nil
 		go conn.Close()
 	}
-	r.mu.Unlock()
-	return nil
 }