Advance the tracked snapshot version while applying live state events

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 {

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