From 491677db6573c196df46f0934e03ea708e17050b Mon Sep 17 00:00:00 2001 From: Vipul Subhash Pandey Date: Wed, 5 Aug 2026 18:28:47 +0530 Subject: [PATCH 1/2] Advance the tracked snapshot version while applying live state events PatchKubernetesApplicationLiveState fetches the stored snapshot once, then loops over a batch of events checking each one against snapshot.Version to skip stale events. The loop never updates snapshot.Version after applying an event, so every event in the batch is compared against the same original version instead of the version the previous event in the loop just brought the state to. Practical effect: if a batch contains an older event for a resource arriving after a newer one for the same resource (a redelivered retry is a normal way this happens), the staleness check does not catch it, and the older event silently overwrites the newer state. No error is logged. Fix: after applying an event, set snapshot.Version to that event's version, so any older/duplicate event later in the same batch is correctly rejected. Added a test that reproduces this with two events for the same resource delivered out of order; it fails on the current code and passes with the fix. Signed-off-by: Vipul Subhash Pandey --- .../server/applicationlivestatestore/store.go | 5 ++ .../applicationlivestatestore/store_test.go | 67 +++++++++++++++++++ 2 files changed, 72 insertions(+) diff --git a/pkg/app/server/applicationlivestatestore/store.go b/pkg/app/server/applicationlivestatestore/store.go index c5b067ec06..767e84864c 100644 --- a/pkg/app/server/applicationlivestatestore/store.go +++ b/pkg/app/server/applicationlivestatestore/store.go @@ -117,6 +117,11 @@ func (s *store) PatchKubernetesApplicationLiveState(ctx context.Context, events case model.KubernetesResourceStateEvent_DELETED: snapshot.Kubernetes.Resources = mergeKubernetesResourceStatesOnDeleted(snapshot.Kubernetes.Resources, ev) } + // Advance the in-memory baseline to this event's version so that, within + // this same batch, an older or duplicate event for the same application + // (e.g. redelivered after a retry) is correctly recognized as stale and + // skipped by the check above instead of overwriting the newer state. + snapshot.Version = ev.SnapshotVersion } for applicationID, snapshot := range snapshots { diff --git a/pkg/app/server/applicationlivestatestore/store_test.go b/pkg/app/server/applicationlivestatestore/store_test.go index 9ddad29176..4f03be4fc5 100644 --- a/pkg/app/server/applicationlivestatestore/store_test.go +++ b/pkg/app/server/applicationlivestatestore/store_test.go @@ -15,13 +15,80 @@ package applicationlivestatestore import ( + "context" "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + "github.com/pipe-cd/pipecd/pkg/cache/memorycache" "github.com/pipe-cd/pipecd/pkg/model" ) +// TestPatchKubernetesApplicationLiveState_OutOfOrderEvents proves that a single +// call carrying two events for the same application, where an older event +// happens to come after a newer one (e.g. because it was redelivered by a +// retry), does not let the older event overwrite the state the newer one +// already applied. +// +// This can genuinely happen: the piped-side reporter batches events and +// resends unacknowledged ones, so a duplicate/stale event landing after a +// fresher one for the same resource is not a made-up scenario. +func TestPatchKubernetesApplicationLiveState_OutOfOrderEvents(t *testing.T) { + s := &store{ + cache: &applicationLiveStateCache{backend: memorycache.NewCache()}, + logger: zap.NewNop(), + } + + const appID = "app-1" + baseVersion := model.ApplicationLiveStateVersion{Timestamp: 100, Index: 0} + newerVersion := model.ApplicationLiveStateVersion{Timestamp: 100, Index: 1} + + initial := &model.ApplicationLiveStateSnapshot{ + ApplicationId: appID, + Kind: model.ApplicationKind_KUBERNETES, + Version: &baseVersion, + Kubernetes: &model.KubernetesApplicationLiveState{ + Resources: []*model.KubernetesResourceState{ + {Id: "pod-1", HealthStatus: model.KubernetesResourceState_UNKNOWN}, + }, + }, + } + require.NoError(t, s.cache.Put(appID, initial)) + + // The true, later event: the pod recovered. Its SnapshotVersion is the + // version the app was at right before this change (newerVersion), + // i.e. it is chronologically the second change. + recoveredEvent := &model.KubernetesResourceStateEvent{ + Id: "event-recovered", + ApplicationId: appID, + Type: model.KubernetesResourceStateEvent_ADD_OR_UPDATED, + SnapshotVersion: &newerVersion, + State: &model.KubernetesResourceState{Id: "pod-1", HealthStatus: model.KubernetesResourceState_HEALTHY}, + } + // A stale redelivery of the first change (the crash), arriving in the + // same batch AFTER the recovery event above. + staleCrashEvent := &model.KubernetesResourceStateEvent{ + Id: "event-crash-retry", + ApplicationId: appID, + Type: model.KubernetesResourceStateEvent_ADD_OR_UPDATED, + SnapshotVersion: &baseVersion, + State: &model.KubernetesResourceState{Id: "pod-1", HealthStatus: model.KubernetesResourceState_OTHER}, + } + + s.PatchKubernetesApplicationLiveState(context.Background(), []*model.KubernetesResourceStateEvent{ + recoveredEvent, + staleCrashEvent, + }) + + got, err := s.GetStateSnapshot(context.Background(), appID) + require.NoError(t, err) + require.Len(t, got.Kubernetes.Resources, 1) + assert.Equal(t, model.KubernetesResourceState_HEALTHY, got.Kubernetes.Resources[0].HealthStatus, + "the stale retried event must not overwrite the newer state that was already applied in this same batch") +} + func TestMergeKubernetesResourceStatesOnAddOrUpdated(t *testing.T) { testcases := []struct { name string From d43a57a3e1be993a38234fa766d222e77d491095 Mon Sep 17 00:00:00 2001 From: Vipul Subhash Pandey Date: Sat, 22 Aug 2026 16:04:44 +0530 Subject: [PATCH 2/2] Fix gofmt alignment in store_test.go Signed-off-by: Vipul Subhash Pandey --- pkg/app/server/applicationlivestatestore/store_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/app/server/applicationlivestatestore/store_test.go b/pkg/app/server/applicationlivestatestore/store_test.go index 4f03be4fc5..1a6292d305 100644 --- a/pkg/app/server/applicationlivestatestore/store_test.go +++ b/pkg/app/server/applicationlivestatestore/store_test.go @@ -37,7 +37,7 @@ import ( // fresher one for the same resource is not a made-up scenario. func TestPatchKubernetesApplicationLiveState_OutOfOrderEvents(t *testing.T) { s := &store{ - cache: &applicationLiveStateCache{backend: memorycache.NewCache()}, + cache: &applicationLiveStateCache{backend: memorycache.NewCache()}, logger: zap.NewNop(), }