From 12aa0a16dacf6475776466a22e65b7eb85e5f20d Mon Sep 17 00:00:00 2001 From: Janez Podhostnik Date: Mon, 3 Aug 2026 16:17:16 +0200 Subject: [PATCH] Add lifecycle regression tests for grpcserver and optimistic sync pipeline --- .../optimistic_sync/pipeline/pipeline_test.go | 36 ++++++++ module/grpcserver/server_test.go | 83 +++++++++++++++++++ 2 files changed, 119 insertions(+) create mode 100644 module/grpcserver/server_test.go diff --git a/module/executiondatasync/optimistic_sync/pipeline/pipeline_test.go b/module/executiondatasync/optimistic_sync/pipeline/pipeline_test.go index ea2ef161ce1..ae8c531cd13 100644 --- a/module/executiondatasync/optimistic_sync/pipeline/pipeline_test.go +++ b/module/executiondatasync/optimistic_sync/pipeline/pipeline_test.go @@ -16,6 +16,42 @@ import ( "github.com/onflow/flow-go/utils/unittest" ) +// TestPipelineParentUpdateBeforeRunNotLost is a regression test verifying that a parent state +// update delivered before (or concurrently with) Run is not lost. Run is invoked with a stale +// initial parent state (StatePending), simulating a parent update that arrives while the +// pipeline is being started. Before the fix, Run unconditionally overwrote the cached parent +// state with the stale initial value, so the pipeline never observed the parent reaching +// StateComplete and deadlocked in StateWaitingPersist without ever persisting. +func TestPipelineParentUpdateBeforeRunNotLost(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + pipeline, mockCore, updateChan, _ := createPipeline(t) + + mockCore.On("Download", mock.Anything).Return(nil) + mockCore.On("Index").Return(nil) + mockCore.On("Persist").Return(nil) + + // the parent completes and the result is sealed BEFORE Run is called + pipeline.OnParentStateUpdated(optimistic_sync.StateComplete) + pipeline.SetSealed() + + go func() { + // Run receives the (now stale) initial parent state + err := pipeline.Run(context.Background(), mockCore, optimistic_sync.StatePending) + require.NoError(t, err) + }() + + // despite the stale initial state, the pipeline must progress all the way to Complete + for _, expected := range []optimistic_sync.State{optimistic_sync.StateProcessing, optimistic_sync.StateWaitingPersist, optimistic_sync.StateComplete} { + synctest.Wait() + assertUpdate(t, updateChan, expected) + } + + // wait for Run goroutine to finish + synctest.Wait() + assertNoUpdate(t, pipeline, updateChan, optimistic_sync.StateComplete) + }) +} + // TestPipelineStateTransitions verifies that the pipeline correctly transitions // through states when provided with the correct conditions. func TestPipelineStateTransitions(t *testing.T) { diff --git a/module/grpcserver/server_test.go b/module/grpcserver/server_test.go new file mode 100644 index 00000000000..0bd993e3826 --- /dev/null +++ b/module/grpcserver/server_test.go @@ -0,0 +1,83 @@ +package grpcserver_test + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.uber.org/atomic" + "google.golang.org/grpc" + + "github.com/onflow/flow-go/module/grpcserver" + "github.com/onflow/flow-go/module/irrecoverable" + "github.com/onflow/flow-go/utils/unittest" +) + +func serverFixture(t *testing.T, listenAddr string) *grpcserver.GrpcServer { + return grpcserver.NewGrpcServer( + unittest.Logger(), + listenAddr, + grpc.NewServer(), + atomic.NewPointer[irrecoverable.SignalerContext](nil), + ) +} + +// TestGrpcServer_StartStop verifies a normal server lifecycle: the server starts, becomes +// ready, and shuts down without throwing an irrecoverable error. +func TestGrpcServer_StartStop(t *testing.T) { + server := serverFixture(t, "localhost:0") + + ctx, cancel := context.WithCancel(context.Background()) + signalerCtx := irrecoverable.NewMockSignalerContext(t, ctx) // fails the test on any Throw + + server.Start(signalerCtx) + unittest.RequireCloseBefore(t, server.Ready(), 5*time.Second, "server did not start on time") + require.NotNil(t, server.GRPCAddress()) + + cancel() + unittest.RequireCloseBefore(t, server.Done(), 5*time.Second, "server did not stop on time") +} + +// TestGrpcServer_ImmediateShutdown is a regression test for the shutdown race between the +// server's two workers: if the shutdown worker completes GracefulStop before the serve worker +// reaches Serve, Serve returns ErrServerStopped. This is a normal shutdown, and must NOT be +// thrown as an irrecoverable error. The mock signaler context fails the test on any Throw; +// repeated immediate shutdowns make the race likely enough to be exercised. +func TestGrpcServer_ImmediateShutdown(t *testing.T) { + for i := 0; i < 50; i++ { + server := serverFixture(t, "localhost:0") + + ctx, cancel := context.WithCancel(context.Background()) + signalerCtx := irrecoverable.NewMockSignalerContext(t, ctx) // fails the test on any Throw + + server.Start(signalerCtx) + // cancel without waiting for the server to become ready, so that the shutdown races + // the startup + cancel() + unittest.RequireCloseBefore(t, server.Done(), 5*time.Second, "server did not stop on time") + } +} + +// TestGrpcServer_ListenErrorThrown verifies that a genuine startup failure (the listen address +// cannot be bound) is still thrown as an irrecoverable error. +func TestGrpcServer_ListenErrorThrown(t *testing.T) { + server := serverFixture(t, "invalid-listen-address") + + thrown := make(chan error, 1) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + signalerCtx := irrecoverable.NewMockSignalerContextWithCallback(t, ctx, func(err error) { + select { + case thrown <- err: + default: + } + }) + + server.Start(signalerCtx) + + unittest.RequireReturnsBefore(t, func() { + err := <-thrown + require.Error(t, err) + }, 5*time.Second, "expected listen error was not thrown") +}