feat: add support for optimistic concurrency check (#1133)

Signed-off-by: Miguel <[email protected]>

Author: migmartri

app/controlplane/api/controlplane/v1/attestation_state.pb.go

@@ -17,18 +17,19 @@ syntax = "proto3";
 
 package controlplane.v1;
 
-option go_package = "github.com/chainloop-dev/chainloop/app/controlplane/api/controlplane/v1;v1";
-
-import "buf/validate/validate.proto";
 import "attestation/v1/crafting_state.proto";
+import "buf/validate/validate.proto";
+import "errors/errors.proto";
+
+option go_package = "github.com/chainloop-dev/chainloop/app/controlplane/api/controlplane/v1;v1";
 
 // API to remotely store and retrieve attestation state
 // using the attestation crafting process
 service AttestationStateService {
-	rpc Initialized (AttestationStateServiceInitializedRequest) returns (AttestationStateServiceInitializedResponse);
-	rpc Save (AttestationStateServiceSaveRequest) returns (AttestationStateServiceSaveResponse);
-	rpc Read (AttestationStateServiceReadRequest) returns (AttestationStateServiceReadResponse);
-	rpc Reset (AttestationStateServiceResetRequest) returns (AttestationStateServiceResetResponse);
+  rpc Initialized(AttestationStateServiceInitializedRequest) returns (AttestationStateServiceInitializedResponse);
+  rpc Save(AttestationStateServiceSaveRequest) returns (AttestationStateServiceSaveResponse);
+  rpc Read(AttestationStateServiceReadRequest) returns (AttestationStateServiceReadResponse);
+  rpc Reset(AttestationStateServiceResetRequest) returns (AttestationStateServiceResetResponse);
 }
 
 message AttestationStateServiceInitializedRequest {
@@ -37,7 +38,7 @@ message AttestationStateServiceInitializedRequest {
 
 message AttestationStateServiceInitializedResponse {
   Result result = 1;
-  
+
   message Result {
     bool initialized = 1;
   }
@@ -46,6 +47,9 @@ message AttestationStateServiceInitializedResponse {
 message AttestationStateServiceSaveRequest {
   string workflow_run_id = 1 [(buf.validate.field).string = {min_len: 1}];
   attestation.v1.CraftingState attestation_state = 2 [(buf.validate.field).required = true];
+  // digest of the attestation state this update was performed on top of
+  // The digest might be empty the first time
+  string base_digest = 3;
 }
 
 message AttestationStateServiceSaveResponse {}
@@ -56,9 +60,11 @@ message AttestationStateServiceReadRequest {
 
 message AttestationStateServiceReadResponse {
   Result result = 1;
-  
+
   message Result {
     attestation.v1.CraftingState attestation_state = 2;
+    // digest of the attestation state to implement Optimistic Concurrency Control
+    string digest = 3;
   }
 }
 
@@ -68,3 +74,8 @@ message AttestationStateServiceResetRequest {
 
 message AttestationStateServiceResetResponse {}
 
+enum AttestationStateError {
+  ATTESTATION_STATE_ERROR_UNSPECIFIED = 0;
+  // https://developer.mozilla.org/en-US/docs/Web/HTTP/Status/409
+  ATTESTATION_STATE_ERROR_CONFLICT = 1 [(errors.code) = 409];
+}

app/controlplane/api/controlplane/v1/attestation_state.proto

@@ -93,7 +93,12 @@ func (s *AttestationStateService) Save(ctx context.Context, req *cpAPI.Attestati
 		return nil, errors.Forbidden("forbidden", "failed to authenticate request")
 	}
 
-	if err := s.attestationStateUseCase.Save(ctx, wf.ID.String(), req.WorkflowRunId, req.AttestationState, encryptionPassphrase); err != nil {
+	err = s.attestationStateUseCase.Save(ctx, wf.ID.String(), req.WorkflowRunId, req.AttestationState, encryptionPassphrase, biz.WithAttStateBaseDigest(req.GetBaseDigest()))
+	if err != nil {
+		if biz.IsErrAttestationStateConflict(err) {
+			return nil, cpAPI.ErrorAttestationStateErrorConflict(err.Error())
+		}
+
 		return nil, handleUseCaseErr(err, s.log)
 	}
 
@@ -124,6 +129,7 @@ func (s *AttestationStateService) Read(ctx context.Context, req *cpAPI.Attestati
 	return &cpAPI.AttestationStateServiceReadResponse{
 		Result: &cpAPI.AttestationStateServiceReadResponse_Result{
 			AttestationState: state.State,
+			Digest:           state.Digest,
 		},
 	}, nil
 }

app/controlplane/api/controlplane/v1/attestation_state_errors.pb.go

@@ -33,12 +33,14 @@ import (
 
 type AttestationState struct {
 	State *v1.CraftingState
+	// Digest will be used for optimistic concurrency control
+	Digest string
 }
 
 type AttestationStateRepo interface {
 	Initialized(ctx context.Context, workflowRunID uuid.UUID) (bool, error)
-	Save(ctx context.Context, workflowRunID uuid.UUID, state []byte) error
-	Read(ctx context.Context, workflowRunID uuid.UUID) ([]byte, error)
+	Save(ctx context.Context, workflowRunID uuid.UUID, state []byte, baseDigest string) error
+	Read(ctx context.Context, workflowRunID uuid.UUID) ([]byte, string, error)
 	Reset(ctx context.Context, workflowRunID uuid.UUID) error
 }
 
@@ -65,7 +67,25 @@ func (uc *AttestationStateUseCase) Initialized(ctx context.Context, workflowID,
 	return initialized, nil
 }
 
-func (uc *AttestationStateUseCase) Save(ctx context.Context, workflowID, runID string, state *v1.CraftingState, passphrase string) error {
+type AttestationStateSaveOpts struct {
+	BaseDigest string
+}
+
+type SaveOption func(*AttestationStateSaveOpts)
+
+func WithAttStateBaseDigest(digest string) SaveOption {
+	return func(o *AttestationStateSaveOpts) {
+		o.BaseDigest = digest
+	}
+}
+
+func (uc *AttestationStateUseCase) Save(ctx context.Context, workflowID, runID string, state *v1.CraftingState, passphrase string, opts ...SaveOption) error {
+	opt := &AttestationStateSaveOpts{}
+
+	for _, o := range opts {
+		o(opt)
+	}
+
 	runUUID, err := uc.checkWorkflowRunInWorkflow(ctx, workflowID, runID)
 	if err != nil {
 		return fmt.Errorf("failed to check workflow run: %w", err)
@@ -76,12 +96,16 @@ func (uc *AttestationStateUseCase) Save(ctx context.Context, workflowID, runID s
 		return fmt.Errorf("failed to marshal attestation state: %w", err)
 	}
 
+	if passphrase == "" {
+		return NewErrValidationStr("passphrase is required")
+	}
+
 	encryptedState, err := encrypt(rawState, passphrase)
 	if err != nil {
 		return fmt.Errorf("failed to encrypt attestation state: %w", err)
 	}
 
-	if err := uc.repo.Save(ctx, *runUUID, encryptedState); err != nil {
+	if err := uc.repo.Save(ctx, *runUUID, encryptedState, opt.BaseDigest); err != nil {
 		return fmt.Errorf("failed to save attestation state: %w", err)
 	}
 
@@ -94,7 +118,7 @@ func (uc *AttestationStateUseCase) Read(ctx context.Context, workflowID, runID,
 		return nil, fmt.Errorf("failed to check workflow run: %w", err)
 	}
 
-	res, err := uc.repo.Read(ctx, *runUUID)
+	res, digest, err := uc.repo.Read(ctx, *runUUID)
 	if err != nil {
 		return nil, fmt.Errorf("failed to read attestation state: %w", err)
 	}
@@ -109,7 +133,7 @@ func (uc *AttestationStateUseCase) Read(ctx context.Context, workflowID, runID,
 		return nil, fmt.Errorf("failed to unmarshal attestation state: %w", err)
 	}
 
-	return &AttestationState{State: state}, nil
+	return &AttestationState{State: state, Digest: digest}, nil
 }
 
 func (uc *AttestationStateUseCase) Reset(ctx context.Context, workflowID, runID string) error {