refactor(app): drop the no-op broker ready reconnect flag

commit: 357815cd16faba89c27d0653a5c6212d541966e4
author: brice · 29d ago · 2026-09-23 14:45 UTC
committer: brice · 29d ago · 2026-09-23 14:49 UTC
parent: d98363f
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +9 -14

M internal/app/broker_ready.go +8 -11


@@ -143,22 +143,19 @@ func runBrokerReady(ctx context.Context, o brokerReadyOptions, out io.Writer) er
143	143		}
144	144		continue
145	145		}
146		-	result, reconnect := observeBrokerReady(deadline, o, service)
	146	+	result := observeBrokerReady(deadline, o, service)
147	147		_ = service.Close()
148	148		if result != nil {
149	149		return emitReady(out, *result)
150	150		}
151		-	if !reconnect {
152		-	continue
153		-	}
154	151		}
155	152		}
156	153
157		-	func observeBrokerReady(ctx context.Context, o brokerReadyOptions, service ports.BrokerService) (*brokerReadyResult, bool) {
	154	+	func observeBrokerReady(ctx context.Context, o brokerReadyOptions, service ports.BrokerService) *brokerReadyResult {
158	155		sub, err := service.Subscribe()
159	156		if err != nil {
160	157		r := readyResult(o, "terminal", "protocol", nil)
161		-	return &r, false
	158	+	return &r
162	159		}
163	160		defer sub.Close()
164	161		for {

@@ -166,7 +163,7 @@ func observeBrokerReady(ctx context.Context, o brokerReadyOptions, service ports
166	163		if snapshot.Epoch != 0 {
167	164		if snapshot.Validate() != nil {
168	165		r := readyResult(o, "terminal", "protocol", nil)
169		-	return &r, false
	166	+	return &r
170	167		}
171	168		locals := make([]ports.BrokerDaemonObservation, 0, 1)
172	169		for _, d := range snapshot.Daemons {

@@ -176,20 +173,20 @@ func observeBrokerReady(ctx context.Context, o brokerReadyOptions, service ports
176	173		}
177	174		if len(locals) > 1 {
178	175		r := readyResult(o, "terminal", "protocol", nil)
179		-	return &r, false
	176	+	return &r
180	177		}
181	178		if len(locals) == 1 {
182	179		r := evaluateReady(o, snapshot, locals[0])
183	180		if r.Status == "ready" \|\| r.Status == "terminal" {
184		-	return &r, false
	181	+	return &r
185	182		}
186	183		}
187	184		}
188	185		select {
189	186		case <-ctx.Done():
190		-	return nil, false
	187	+	return nil
191	188		case <-service.Done():
192		-	return nil, true
	189	+	return nil
193	190		case <-sub.Changed():
194	191		}
195	192		}

 			}
 			continue
 		}
-		result, reconnect := observeBrokerReady(deadline, o, service)
+		result := observeBrokerReady(deadline, o, service)
 		_ = service.Close()
 		if result != nil {
 			return emitReady(out, *result)
 		}
-		if !reconnect {
-			continue
-		}
 	}
 }
-func observeBrokerReady(ctx context.Context, o brokerReadyOptions, service ports.BrokerService) (*brokerReadyResult, bool) {
+func observeBrokerReady(ctx context.Context, o brokerReadyOptions, service ports.BrokerService) *brokerReadyResult {
 	sub, err := service.Subscribe()
 	if err != nil {
 		r := readyResult(o, "terminal", "protocol", nil)
-		return &r, false
+		return &r
 	}
 	defer sub.Close()
 	for {
 		if snapshot.Epoch != 0 {
 			if snapshot.Validate() != nil {
 				r := readyResult(o, "terminal", "protocol", nil)
-				return &r, false
+				return &r
 			}
 			locals := make([]ports.BrokerDaemonObservation, 0, 1)
 			for _, d := range snapshot.Daemons {
 			}
 			if len(locals) > 1 {
 				r := readyResult(o, "terminal", "protocol", nil)
-				return &r, false
+				return &r
 			}
 			if len(locals) == 1 {
 				r := evaluateReady(o, snapshot, locals[0])
 				if r.Status == "ready" || r.Status == "terminal" {
-					return &r, false
+					return &r
 				}
 			}
 		}
 		select {
 		case <-ctx.Done():
-			return nil, false
+			return nil
 		case <-service.Done():
-			return nil, true
+			return nil
 		case <-sub.Changed():
 		}
 	}

M internal/app/broker_ready_test.go +1 -3


@@ -590,9 +590,7 @@ func TestBrokerReadyObservationTable(t *testing.T) {
590	590		ctx, cancel := context.WithTimeout(context.Background(), readyTestObserveMax)
591	591		defer cancel()
592	592
593		-	result, reconnect := observeBrokerReady(ctx, brokerReadyOptions{require: tt.require, timeout: time.Second, maxAge: time.Second}, service)
594		-
595		-	require.False(t, reconnect, "a terminal or satisfied observation never reconnects")
	593	+	result := observeBrokerReady(ctx, brokerReadyOptions{require: tt.require, timeout: time.Second, maxAge: time.Second}, service)
596	594		require.Empty(t, service.mutationNames(), "readiness must never open a stream, reconcile, or mutate membership")
597	595		if tt.subErr != nil {
598	596		require.Zero(t, service.subscriptionCloses(), "a refused subscription is never closed as if it existed")


@@ -590,9 +590,7 @@ func TestBrokerReadyObservationTable(t *testing.T) {
590		ctx, cancel := context.WithTimeout(context.Background(), readyTestObserveMax)	590		ctx, cancel := context.WithTimeout(context.Background(), readyTestObserveMax)
591		defer cancel()	591		defer cancel()
592			592
593	-	result, reconnect := observeBrokerReady(ctx, brokerReadyOptions{require: tt.require, timeout: time.Second, maxAge: time.Second}, service)	593	+	result := observeBrokerReady(ctx, brokerReadyOptions{require: tt.require, timeout: time.Second, maxAge: time.Second}, service)
594	-
595	-	require.False(t, reconnect, "a terminal or satisfied observation never reconnects")
596		require.Empty(t, service.mutationNames(), "readiness must never open a stream, reconcile, or mutate membership")	594		require.Empty(t, service.mutationNames(), "readiness must never open a stream, reconcile, or mutate membership")
597		if tt.subErr != nil {	595		if tt.subErr != nil {
598		require.Zero(t, service.subscriptionCloses(), "a refused subscription is never closed as if it existed")	596		require.Zero(t, service.subscriptionCloses(), "a refused subscription is never closed as if it existed")