make the spec update and status update independently (#389)

Signed-off-by: Wei Liu <liuweixa@redhat.com>

Author: skeeey

Makefile

@@ -245,7 +245,7 @@ test-integration-mqtt:
 .PHONY: test-integration-mqtt
 
 test-integration-grpc:
-	BROKER=grpc MAESTRO_ENV=testing gotestsum --jsonfile-timing-events=$(grpc_integration_test_json_output) --format $(TEST_SUMMARY_FORMAT) -- -p 1 -ldflags -s -v -timeout 1h $(TESTFLAGS) \
+	BROKER=grpc MAESTRO_ENV=testing gotestsum --jsonfile-timing-events=$(grpc_integration_test_json_output) --format $(TEST_SUMMARY_FORMAT) -- -count=1 -p 1 -ldflags -s -v -timeout 1h $(TESTFLAGS) \
 			./test/integration
 .PHONY: test-integration-grpc
 
@@ -428,7 +428,7 @@ db/teardown:
 
 .PHONY: mqtt/prepare
 mqtt/prepare:
-	@echo $(shell LC_CTYPE=C tr -dc 'a-zA-Z0-9' < /dev/urandom | head -c 13) > $(mqtt_password_file)
+	@openssl rand -base64 13 | tr -dc 'a-zA-Z0-9' | head -c 13 > $(mqtt_password_file)
 
 .PHONY: mqtt/setup
 mqtt/setup: mqtt/prepare

pkg/dao/mocks/resource.go

@@ -39,6 +39,10 @@ func (d *resourceDaoMock) Update(ctx context.Context, resource *api.Resource) (*
 	return nil, errors.NotImplemented("Resource").AsError()
 }
 
+func (d *resourceDaoMock) UpdateStatus(ctx context.Context, resource *api.Resource) (*api.Resource, error) {
+	return nil, errors.NotImplemented("Resource").AsError()
+}
+
 func (d *resourceDaoMock) Delete(ctx context.Context, id string, unscoped bool) error {
 	return errors.NotImplemented("Resource").AsError()
 }

pkg/dao/resource.go

@@ -13,6 +13,7 @@ type ResourceDao interface {
 	Get(ctx context.Context, id string) (*api.Resource, error)
 	Create(ctx context.Context, resource *api.Resource) (*api.Resource, error)
 	Update(ctx context.Context, resource *api.Resource) (*api.Resource, error)
+	UpdateStatus(ctx context.Context, resource *api.Resource) (*api.Resource, error)
 	Delete(ctx context.Context, id string, unscoped bool) error
 	FindByIDs(ctx context.Context, ids []string) (api.ResourceList, error)
 	FindBySource(ctx context.Context, source string) (api.ResourceList, error)
@@ -51,7 +52,27 @@ func (d *sqlResourceDao) Create(ctx context.Context, resource *api.Resource) (*a
 
 func (d *sqlResourceDao) Update(ctx context.Context, resource *api.Resource) (*api.Resource, error) {
 	g2 := (*d.sessionFactory).New(ctx)
-	if err := g2.Unscoped().Omit(clause.Associations).Updates(resource).Error; err != nil {
+	if err := g2.Unscoped().Omit(clause.Associations).
+		Where("id = ?", resource.ID).
+		Select("version", "payload").
+		Updates(api.Resource{
+			Version: resource.Version,
+			Payload: resource.Payload,
+		}).Error; err != nil {
+		db.MarkForRollback(ctx, err)
+		return nil, err
+	}
+	return resource, nil
+}
+
+func (d *sqlResourceDao) UpdateStatus(ctx context.Context, resource *api.Resource) (*api.Resource, error) {
+	g2 := (*d.sessionFactory).New(ctx)
+	if err := g2.Unscoped().Omit(clause.Associations).
+		Where("id = ?", resource.ID).
+		Select("status").
+		Updates(api.Resource{
+			Status: resource.Status,
+		}).Error; err != nil {
 		db.MarkForRollback(ctx, err)
 		return nil, err
 	}

pkg/services/resource.go

@@ -177,6 +177,9 @@ func (s *sqlResourceService) UpdateStatus(ctx context.Context, resource *api.Res
 	}
 
 	// Make sure the requested resource version is consistent with its database version.
+	// If they do not match, the event is stale and can be safely ignored.
+	// The manifestwork status reflects observed generations, ensuring eventual consistency
+	// through subsequent up-to-date events.
 	if found.Version != resource.Version {
 		log.Warnf("Updating status for stale resource; disregard it: id=%s, foundVersion=%d, wantedVersion=%d",
 			resource.ID, found.Version, resource.Version)
@@ -224,8 +227,9 @@ func (s *sqlResourceService) UpdateStatus(ctx context.Context, resource *api.Res
 		return found, false, nil
 	}
 
+	// Only update resource status
 	found.Status = resource.Status
-	updated, err := s.resourceDao.Update(ctx, found)
+	updated, err := s.resourceDao.UpdateStatus(ctx, found)
 	if err != nil {
 		return nil, false, handleUpdateError("Resource", err)
 	}

test/integration/resource_test.go

@@ -490,7 +490,7 @@ func TestMarkAsDeletingThenUpdate(t *testing.T) {
 			ID: resource.ID,
 		},
 		Version: resource.Version,
-		Status:  createStatusWithSequenceID(t, h, resource.ID, fmt.Sprintf("%d", 1)),
+		Status:  createStatusWithSequenceID(t, resource.ID, fmt.Sprintf("%d", 1)),
 	}
 	_, updated, svcErr := resourceService.UpdateStatus(ctx, statusRes)
 	Expect(svcErr).NotTo(HaveOccurred())
@@ -529,8 +529,118 @@ func updateWorkStatus(ctx context.Context, workClient workv1client.ManifestWorkI
 	return nil
 }
 
+// TestUpdateAndUpdateStatusIsolation ensures that Update and UpdateStatus operations
+// work independently without affecting each other. Specifically:
+// 1. Create a resource
+// 2. Update the status
+// 3. Update the resource payload
+// 4. Verify the status from step 2 is preserved and not affected by the payload update
+func TestUpdateAndUpdateStatusIsolation(t *testing.T) {
+	h, _ := test.RegisterIntegration(t)
+
+	account := h.NewRandAccount()
+	ctx := h.NewAuthenticatedContext(account)
+
+	// Step 1: Create a resource
+	consumer, err := h.CreateConsumer("cluster-" + rand.String(5))
+	Expect(err).NotTo(HaveOccurred())
+	deployName := fmt.Sprintf("nginx-%s", rand.String(5))
+	resource, err := h.CreateResource(uuid.NewString(), consumer.Name, deployName, "default", 1)
+	Expect(err).NotTo(HaveOccurred())
+	Expect(resource.Version).To(Equal(int32(1)))
+	Expect(len(resource.Status)).To(Equal(0), "Initial resource should have no status")
+
+	resourceService := h.Env().Services.Resources()
+
+	// Step 2: Update the status with sequence ID "1"
+	statusRes1 := &api.Resource{
+		Meta: api.Meta{
+			ID: resource.ID,
+		},
+		Version: resource.Version,
+		Status:  createStatusWithSequenceID(t, resource.ID, "1"),
+	}
+	_, updated, svcErr := resourceService.UpdateStatus(ctx, statusRes1)
+	Expect(svcErr).NotTo(HaveOccurred())
+	Expect(updated).Should(BeTrue(), "Status should be updated")
+
+	// Verify status was set correctly
+	updatedRes, svcErr := resourceService.Get(ctx, resource.ID)
+	Expect(svcErr).NotTo(HaveOccurred())
+	Expect(len(updatedRes.Status)).ShouldNot(Equal(0), "Status should not be empty after update")
+	statusEvt, err := api.JSONMAPToCloudEvent(updatedRes.Status)
+	Expect(err).NotTo(HaveOccurred())
+	statusPayload := &workpayload.ManifestBundleStatus{}
+	Expect(statusEvt.DataAs(statusPayload)).NotTo(HaveOccurred())
+	Expect(statusPayload.Conditions).To(HaveLen(1))
+	Expect(statusPayload.Conditions[0].Type).To(Equal("Applied"))
+	Expect(statusPayload.Conditions[0].Status).To(Equal(metav1.ConditionStatus("True")))
+	initialStatusMessage := statusPayload.Conditions[0].Message
+
+	// Version should still be 1 (UpdateStatus doesn't increment version)
+	Expect(updatedRes.Version).To(Equal(int32(1)))
+
+	// Step 3: Update the resource payload (change replicas from 1 to 3)
+	newResource, err := h.NewResource(resource.ID, consumer.Name, deployName, "default", 3, updatedRes.Version)
+	Expect(err).NotTo(HaveOccurred())
+	newResource.ID = updatedRes.ID
+	newResource.Status = createStatusWithSequenceID(t, resource.ID, "2")
+
+	updatedPayloadRes, svcErr := resourceService.Update(ctx, newResource)
+	Expect(svcErr).NotTo(HaveOccurred())
+	Expect(updatedPayloadRes.Version).To(Equal(int32(2)), "Version should be incremented after Update")
+
+	// Step 4: Verify that the status is preserved and unchanged
+	finalRes, svcErr := resourceService.Get(ctx, resource.ID)
+	Expect(svcErr).NotTo(HaveOccurred())
+
+	// Status should still be present and unchanged
+	Expect(len(finalRes.Status)).ShouldNot(Equal(0), "Status should be preserved after Update")
+
+	finalStatusEvt, err := api.JSONMAPToCloudEvent(finalRes.Status)
+	Expect(err).NotTo(HaveOccurred())
+	finalStatusPayload := &workpayload.ManifestBundleStatus{}
+	Expect(finalStatusEvt.DataAs(finalStatusPayload)).NotTo(HaveOccurred())
+	Expect(finalStatusPayload.Conditions).To(HaveLen(1))
+	Expect(finalStatusPayload.Conditions[0].Type).To(Equal("Applied"))
+	Expect(finalStatusPayload.Conditions[0].Status).To(Equal(metav1.ConditionStatus("True")))
+	Expect(finalStatusPayload.Conditions[0].Message).To(Equal(initialStatusMessage), "Status message should be unchanged")
+
+	// Verify the payload was actually updated (replicas changed from 1 to 3)
+	payloadEvt, err := api.JSONMAPToCloudEvent(finalRes.Payload)
+	Expect(err).NotTo(HaveOccurred())
+	payloadData := &workpayload.ManifestBundle{}
+	Expect(payloadEvt.DataAs(payloadData)).NotTo(HaveOccurred())
+	Expect(payloadData.Manifests).To(HaveLen(1))
+
+	var manifest map[string]interface{}
+	Expect(json.Unmarshal(payloadData.Manifests[0].Raw, &manifest)).NotTo(HaveOccurred())
+	spec := manifest["spec"].(map[string]interface{})
+	Expect(spec["replicas"]).To(Equal(float64(3)), "Replicas should be updated to 3")
+
+	// Additional test: Update status again to verify it still works after a payload update
+	statusRes2 := &api.Resource{
+		Meta: api.Meta{
+			ID: resource.ID,
+		},
+		Version: finalRes.Version, // Now version 2
+		Status:  createStatusWithSequenceID(t, resource.ID, "2"),
+	}
+	updatedRes2, updated2, svcErr := resourceService.UpdateStatus(ctx, statusRes2)
+	Expect(svcErr).NotTo(HaveOccurred())
+	Expect(updated2).Should(BeTrue(), "Status should be updated again")
+	Expect(updatedRes2.Version).To(Equal(int32(2)), "Version should remain 2 after UpdateStatus")
+
+	// Verify the new status
+	newStatusEvt, err := api.JSONMAPToCloudEvent(updatedRes2.Status)
+	Expect(err).NotTo(HaveOccurred())
+	newStatusPayload := &workpayload.ManifestBundleStatus{}
+	Expect(newStatusEvt.DataAs(newStatusPayload)).NotTo(HaveOccurred())
+	Expect(newStatusPayload.Conditions[0].Message).To(ContainSubstring("sequence 2"), "Status should reflect new update")
+}
+
 // createStatusWithSequenceID creates a resource status CloudEvent with the given sequence ID
-func createStatusWithSequenceID(t *testing.T, h *test.Helper, resourceID, sequenceID string) map[string]interface{} {
+func createStatusWithSequenceID(t *testing.T, resourceID, sequenceID string) map[string]interface{} {
 	source := "test-agent"
 	eventType := cetypes.CloudEventsType{
 		CloudEventsDataType: workpayload.ManifestBundleEventDataType,