Function TestReceive_ProtocolErrorStreamReconnect

flow/client/client_test.go:452–549

(t *testing.T)

}

func TestReceive_ProtocolErrorStreamReconnect(t *testing.T) {
	server := newTestServer(t)

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	t.Cleanup(cancel)

	client, err := flow.NewClient("http://"+server.addr, "test-payload", "test-signature", 1*time.Second)
	require.NoError(t, err)

	// Cleanups run LIFO: the goroutine-drain registered here runs after Close below,
	// which is when Receive has actually returned. Without this, the Receive goroutine
	// can outlive the test and call t.Logf after teardown, panicking.
	receiveDone := make(chan struct{})
	t.Cleanup(func() {
		select {
		case <-receiveDone:
		case <-time.After(2 * time.Second):
			t.Error("Receive goroutine did not exit after Close")
		}
	})
	t.Cleanup(func() {
		err := client.Close()
		assert.NoError(t, err, "failed to close flow")
	})

	// Track acks received before and after server-side stream close
	var ackCount atomic.Int32
	receivedFirst := make(chan struct{})
	receivedAfterReconnect := make(chan struct{})

	go func() {
		defer close(receiveDone)
		err := client.Receive(ctx, 1*time.Second, func(msg *proto.FlowEventAck) error {
			if msg.IsInitiator || len(msg.EventId) == 0 {
				return nil
			}
			n := ackCount.Add(1)
			if n == 1 {
				close(receivedFirst)
			}
			if n == 2 {
				close(receivedAfterReconnect)
			}
			return nil
		})
		if err != nil && !errors.Is(err, context.Canceled) {
			t.Logf("receive error: %v", err)
		}
	}()

	// Wait for stream to be established, then send first ack
	select {
	case <-server.handlerStarted:
	case <-time.After(3 * time.Second):
		t.Fatal("timeout waiting for stream to be established")
	}
	server.acks <- &proto.FlowEventAck{EventId: []byte("before-close")}