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..1a6292d305 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