fix(apps): start zero-replica stopped components

controllers/apps/component/transformer_component_workload.go

@@ -294,7 +294,6 @@ func (t *componentWorkloadTransformer) startWorkload(
 	}
 
 	delete(protoITS.Annotations, stopReplicasSnapshotKey)
-	delete(runningITS.Annotations, stopReplicasSnapshotKey)
 
 	return nil
 }
@@ -334,6 +333,13 @@ func copyAndMergeITS(oldITS, newITS *workloads.InstanceSet) *workloads.InstanceS
 		})
 	}
 	intctrlutil.MergeMetadataMapInplace(itsProto.Annotations, &itsObjCopy.Annotations)
+	// The stop snapshot is owned by the component controller. Its absence from
+	// the desired object means that the workload has been started and the
+	// persisted annotation must be removed. Do this on the update copy rather
+	// than mutating the informer/cache object in startWorkload.
+	if _, ok := itsProto.Annotations[stopReplicasSnapshotKey]; !ok {
+		delete(itsObjCopy.Annotations, stopReplicasSnapshotKey)
+	}
 	intctrlutil.MergeMetadataMapInplace(itsProto.Labels, &itsObjCopy.Labels)
 	// merge pod spec template annotations
 	intctrlutil.MergeMetadataMapInplace(itsProto.Spec.Template.Annotations, &itsObjCopy.Spec.Template.Annotations)

controllers/apps/component/transformer_component_workload_test.go

@@ -194,4 +194,30 @@ var _ = Describe("Component Workload Operations Test", func() {
 			Expect(ops.leaveMemberForPod(pod1, pods)).Should(Succeed())
 		})
 	})
+
+	Context("start and stop operations", func() {
+		It("removes the stop snapshot when starting a zero-replica workload", func() {
+			const snapshot = `{"":0}`
+			runningITS := testapps.NewInstanceSetFactory(testCtx.DefaultNamespace,
+				"test-its", clusterName, compName).
+				SetReplicas(0).
+				AddAnnotations(stopReplicasSnapshotKey, snapshot).
+				GetObject()
+			protoITS := runningITS.DeepCopy()
+
+			transformer := &componentWorkloadTransformer{}
+			Expect(transformer.startWorkload(synthesizeComp, runningITS, protoITS)).Should(Succeed())
+
+			By("not mutating the informer/cache object")
+			Expect(runningITS.Annotations).Should(HaveKeyWithValue(stopReplicasSnapshotKey, snapshot))
+			Expect(protoITS.Annotations).ShouldNot(HaveKey(stopReplicasSnapshotKey))
+
+			By("producing an update even though the restored replica count remains zero")
+			updatedITS := copyAndMergeITS(runningITS, protoITS)
+			Expect(updatedITS).ShouldNot(BeNil())
+			Expect(updatedITS.Annotations).ShouldNot(HaveKey(stopReplicasSnapshotKey))
+			Expect(updatedITS.Spec.Replicas).ShouldNot(BeNil())
+			Expect(*updatedITS.Spec.Replicas).Should(BeEquivalentTo(0))
+		})
+	})
 })