fix(pipe): flaky test - pico@58ebb03

Commit `58ebb03`

Eric Bower · 2026-08-30 09:36:14 -0400 EDT

parent d8d0754

fix(pipe): flaky test

1 files changed, +32, -6

M pkg/apps/pipe/ssh_test.go

+32, -6

......@@ -2337,7 +2337,25 @@ func TestSSH_WildcardSub(t *testing.T) {
		}
	}()

	time.Sleep(100 * time.Millisecond)
	// Wait for subscriber to be registered in the PubSub broker
	subReady := false
	for i := 0; i < 100; i++ {
		for topic, ch := range server.PipeHandler.PubSub.GetChannels() {
			if topic == "alice/metric-drain*" {
				for range ch.GetClients() {
					subReady = true
					break
				}
			}
		}
		if subReady {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}
	if !subReady {
		t.Fatal("timed out waiting for wildcard subscriber to register")
	}

	// 2. Publish to "metric-drain-pgs"
	pubClient1, err := user.NewClient()

......@@ -2375,12 +2393,20 @@ func TestSSH_WildcardSub(t *testing.T) {
	}()
	_ = pubSession2.Run("pub metric-drain-prose -b=false")

	time.Sleep(150 * time.Millisecond)
	_ = subSession.Close()
	// Wait for both events to be received by the subscriber
	deadline := time.Now().Add(5 * time.Second)
	var output string
	for time.Now().Before(deadline) {
		bufMu.Lock()
		output = buf.String()
		bufMu.Unlock()
		if strings.Contains(output, "pgs-event") && strings.Contains(output, "prose-event") {
			break
		}
		time.Sleep(10 * time.Millisecond)
	}

	bufMu.Lock()
	output := buf.String()
	bufMu.Unlock()
	_ = subSession.Close()

	if !strings.Contains(output, "pgs-event") {
		t.Errorf("expected SSH wildcard subscriber output to contain 'pgs-event', got: %q", output)

...	...		@@ -2337,7 +2337,25 @@ func TestSSH_WildcardSub(t *testing.T) {
2337	2337		}
2338	2338		}()
2339	2339
2340		-	time.Sleep(100 * time.Millisecond)
	2340	+	// Wait for subscriber to be registered in the PubSub broker
	2341	+	subReady := false
	2342	+	for i := 0; i < 100; i++ {
	2343	+	for topic, ch := range server.PipeHandler.PubSub.GetChannels() {
	2344	+	if topic == "alice/metric-drain*" {
	2345	+	for range ch.GetClients() {
	2346	+	subReady = true
	2347	+	break
	2348	+	}
	2349	+	}
	2350	+	}
	2351	+	if subReady {
	2352	+	break
	2353	+	}
	2354	+	time.Sleep(10 * time.Millisecond)
	2355	+	}
	2356	+	if !subReady {
	2357	+	t.Fatal("timed out waiting for wildcard subscriber to register")
	2358	+	}
2341	2359
2342	2360		// 2. Publish to "metric-drain-pgs"
2343	2361		pubClient1, err := user.NewClient()

...	...		@@ -2375,12 +2393,20 @@ func TestSSH_WildcardSub(t *testing.T) {
2375	2393		}()
2376	2394		_ = pubSession2.Run("pub metric-drain-prose -b=false")
2377	2395
2378		-	time.Sleep(150 * time.Millisecond)
2379		-	_ = subSession.Close()
	2396	+	// Wait for both events to be received by the subscriber
	2397	+	deadline := time.Now().Add(5 * time.Second)
	2398	+	var output string
	2399	+	for time.Now().Before(deadline) {
	2400	+	bufMu.Lock()
	2401	+	output = buf.String()
	2402	+	bufMu.Unlock()
	2403	+	if strings.Contains(output, "pgs-event") && strings.Contains(output, "prose-event") {
	2404	+	break
	2405	+	}
	2406	+	time.Sleep(10 * time.Millisecond)
	2407	+	}
2380	2408
2381		-	bufMu.Lock()
2382		-	output := buf.String()
2383		-	bufMu.Unlock()
	2409	+	_ = subSession.Close()
2384	2410
2385	2411		if !strings.Contains(output, "pgs-event") {
2386	2412		t.Errorf("expected SSH wildcard subscriber output to contain 'pgs-event', got: %q", output)

Commit 58ebb03

Commit `58ebb03`