feat(cli): store remote digest in local state for concurrency (#1136)

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

Author: migmartri

app/cli/cmd/attestation_add.go

@@ -19,12 +19,15 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"time"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 	"google.golang.org/grpc"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/chainloop-dev/chainloop/app/cli/internal/action"
+	v1 "github.com/chainloop-dev/chainloop/app/controlplane/api/controlplane/v1"
 	schemaapi "github.com/chainloop-dev/chainloop/app/controlplane/api/workflowcontract/v1"
 )
 
@@ -85,18 +88,34 @@ func newAttestationAddCmd() *cobra.Command {
 				return err
 			}
 
-			if err := a.Run(cmd.Context(), attestationID, name, value, kind, annotations); err != nil {
-				if errors.Is(err, action.ErrAttestationNotInitialized) {
-					return err
-				}
+			// In some cases, the attestation state is stored remotely. To control concurrency we use
+			// optimistic locking. We retry the operation if the state has changed since we last read it.
+			return backoff.RetryNotify(
+				func() error {
+					if err := a.Run(cmd.Context(), attestationID, name, value, kind, annotations); err != nil {
+						if errors.Is(err, action.ErrAttestationNotInitialized) {
+							return err
+						}
 
-				return newGracefulError(err)
-			}
+						// We want to retry if the error is a conflict
+						if v1.IsAttestationStateErrorConflict(err) {
+							return err
+						}
 
-			logger.Info().Msg("material added to attestation")
+						// if it's another kind of error we want to stop retrying
+						return backoff.Permanent(newGracefulError(err))
+					}
 
-			return nil
+					logger.Info().Msg("material added to attestation")
+
+					return nil
+				},
+				backoff.NewExponentialBackOff(backoff.WithMaxElapsedTime(10*time.Second)),
+				func(err error, delay time.Duration) {
+					logger.Err(err).Msgf("retrying in %s", delay)
+				})
 		},
+
 		PostRunE: func(cmd *cobra.Command, args []string) error {
 			if artifactCASConn != nil {
 				return artifactCASConn.Close()

app/cli/internal/action/action.go

@@ -43,9 +43,15 @@ func newCrafter(enableRemoteState bool, conn *grpc.ClientConn, opts ...crafter.N
 	var stateManager crafter.StateManager
 	var err error
 
+	// run opts to extract logger
+	c := &crafter.Crafter{}
+	for _, opt := range opts {
+		_ = opt(c)
+	}
+
 	switch enableRemoteState {
 	case true:
-		stateManager, err = remote.New(pb.NewAttestationStateServiceClient(conn))
+		stateManager, err = remote.New(pb.NewAttestationStateServiceClient(conn), c.Logger)
 	case false:
 		stateManager, err = filesystem.New(filepath.Join(os.TempDir(), "chainloop-attestation.tmp.json"))
 	}

app/cli/internal/action/attestation_status.go

@@ -107,7 +107,7 @@ func (action *AttestationStatus) Run(ctx context.Context, attestationID string)
 	// 1. Populate the materials that are defined in the contract schema
 	// 2. Populate the materials that are not defined in the contract schema, added inline in the attestation
 	// In order to avoid duplicates, we keep track of the visited materials
-	if err := populateMaterials(c.CraftingState, res); err != nil {
+	if err := populateMaterials(c.CraftingState.CraftingState, res); err != nil {
 		return nil, fmt.Errorf("populating materials: %w", err)
 	}
 

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.28.6
 	github.com/aws/aws-sdk-go-v2/service/sso v1.20.4
 	github.com/aws/smithy-go v1.20.2
-	github.com/cenkalti/backoff/v4 v4.2.1
+	github.com/cenkalti/backoff/v4 v4.3.0
 	github.com/coreos/go-oidc/v3 v3.10.0
 	github.com/docker/distribution v2.8.3+incompatible
 	github.com/docker/go-connections v0.4.0

go.sum

@@ -293,8 +293,8 @@ github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QH
 github.com/cenkalti/backoff/v3 v3.2.2 h1:cfUAAO3yvKMYKPrvhDuHSwQnhZNk/RMHKdZqKTxfm6M=
 github.com/cenkalti/backoff/v3 v3.2.2/go.mod h1:cIeZDE3IrqwwJl6VUwCN6trj1oXrTS4rc0ij+ULvLYs=
 github.com/cenkalti/backoff/v4 v4.1.1/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
-github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
-github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
+github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
+github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/census-instrumentation/opencensus-proto v0.3.0/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/census-instrumentation/opencensus-proto v0.4.1 h1:iKLQ0xPNFxR/2hzXZMrBo8f1j86j5WHzznCCQxV/b8g=

internal/attestation/crafter/crafter.go

@@ -47,18 +47,18 @@ type StateManager interface {
 	// Check if the state is already initialized
 	Initialized(ctx context.Context, key string) (bool, error)
 	// Write the state to the manager backend
-	Write(ctx context.Context, key string, state *api.CraftingState) error
+	Write(ctx context.Context, key string, state *VersionedCraftingState) error
 	// Read the state from the manager backend
-	Read(ctx context.Context, key string, state *api.CraftingState) error
+	Read(ctx context.Context, key string, state *VersionedCraftingState) error
 	// Reset/Delete the state
 	Reset(ctx context.Context, key string) error
 	// String returns a string representation of the state manager
 	Info(ctx context.Context, key string) string
 }
 
 type Crafter struct {
-	logger        *zerolog.Logger
-	CraftingState *api.CraftingState
+	Logger        *zerolog.Logger
+	CraftingState *VersionedCraftingState
 	Runner        SupportedRunner
 	workingDir    string
 	stateManager  StateManager
@@ -67,13 +67,19 @@ type Crafter struct {
 	validator       *protovalidate.Validator
 }
 
+type VersionedCraftingState struct {
+	*api.CraftingState
+	// This digest is used to verify the integrity of the state during updates
+	UpdateCheckSum string
+}
+
 var ErrAttestationStateNotLoaded = errors.New("crafting state not loaded")
 
 type NewOpt func(c *Crafter) error
 
 func WithLogger(l *zerolog.Logger) NewOpt {
 	return func(c *Crafter) error {
-		c.logger = l
+		c.Logger = l
 		return nil
 	}
 }
@@ -108,7 +114,7 @@ func NewCrafter(stateManager StateManager, opts ...NewOpt) (*Crafter, error) {
 
 	cw, _ := os.Getwd()
 	c := &Crafter{
-		logger:       &noopLogger,
+		Logger:       &noopLogger,
 		workingDir:   cw,
 		stateManager: stateManager,
 		// By default we authenticate with the current user's keychain (i.e ~/.docker/config.json)
@@ -225,11 +231,13 @@ func (c *Crafter) initCraftingStateFile(
 		return fmt.Errorf("initializing crafting state: %w", err)
 	}
 
-	if err := c.stateManager.Write(ctx, attestationID, state); err != nil {
+	// newState doesn't have a digest to check against
+	newState := &VersionedCraftingState{CraftingState: state}
+	if err := c.stateManager.Write(ctx, attestationID, newState); err != nil {
 		return fmt.Errorf("failed to persist crafting state: %w", err)
 	}
 
-	c.logger.Debug().Str("state", c.stateManager.Info(ctx, attestationID)).Msg("created state file")
+	c.Logger.Debug().Str("state", c.stateManager.Info(ctx, attestationID)).Msg("created state file")
 
 	return c.LoadCraftingState(ctx, attestationID)
 }
@@ -240,9 +248,9 @@ func (c *Crafter) Reset(ctx context.Context, stateID string) error {
 }
 
 func (c *Crafter) LoadCraftingState(ctx context.Context, attestationID string) error {
-	c.logger.Debug().Str("state", c.stateManager.Info(ctx, attestationID)).Msg("loading state")
+	c.Logger.Debug().Str("state", c.stateManager.Info(ctx, attestationID)).Msg("loading state")
 
-	c.CraftingState = &api.CraftingState{}
+	c.CraftingState = &VersionedCraftingState{CraftingState: &api.CraftingState{}}
 
 	if err := c.stateManager.Read(ctx, attestationID, c.CraftingState); err != nil {
 		return fmt.Errorf("failed to load crafting state: %w", err)
@@ -255,7 +263,7 @@ func (c *Crafter) LoadCraftingState(ctx context.Context, attestationID string) e
 	}
 
 	c.Runner = NewRunner(runnerType)
-	c.logger.Debug().Str("state", c.stateManager.Info(ctx, attestationID)).Msg("loaded state")
+	c.Logger.Debug().Str("state", c.stateManager.Info(ctx, attestationID)).Msg("loaded state")
 
 	return nil
 }
@@ -413,7 +421,7 @@ func (c *Crafter) ResolveEnvVars(ctx context.Context, attestationID string) erro
 	}
 
 	// Runner specific environment variables
-	c.logger.Debug().Str("runnerType", c.Runner.ID().String()).Msg("loading runner specific env variables")
+	c.Logger.Debug().Str("runnerType", c.Runner.ID().String()).Msg("loading runner specific env variables")
 	if !c.Runner.CheckEnv() {
 		errorStr := fmt.Sprintf("couldn't detect the environment %q. Is the crafting process happening in the target env?", c.Runner.ID().String())
 		return fmt.Errorf("%s - %w", errorStr, ErrRunnerContextNotFound)
@@ -424,7 +432,7 @@ func (c *Crafter) ResolveEnvVars(ctx context.Context, attestationID string) erro
 	for index, envVarDef := range c.Runner.ListEnvVars() {
 		varNames[index] = envVarDef.Name
 	}
-	c.logger.Debug().Str("runnerType", c.Runner.ID().String()).Strs("variables", varNames).Msg("list of env variables to automatically extract")
+	c.Logger.Debug().Str("runnerType", c.Runner.ID().String()).Strs("variables", varNames).Msg("list of env variables to automatically extract")
 
 	outputEnvVars, errors := c.Runner.ResolveEnvVars()
 	if len(errors) > 0 {
@@ -437,7 +445,7 @@ func (c *Crafter) ResolveEnvVars(ctx context.Context, attestationID string) erro
 
 	// User-defined environment vars
 	if len(c.CraftingState.InputSchema.EnvAllowList) > 0 {
-		c.logger.Debug().Strs("allowList", c.CraftingState.InputSchema.EnvAllowList).Msg("loading env variables")
+		c.Logger.Debug().Strs("allowList", c.CraftingState.InputSchema.EnvAllowList).Msg("loading env variables")
 	}
 	for _, want := range c.CraftingState.InputSchema.EnvAllowList {
 		val := os.Getenv(want)
@@ -501,7 +509,7 @@ func (c *Crafter) AddMaterialFromContract(ctx context.Context, attestationID, ke
 
 	// 2 - Check that it has not been set yet and warn of override
 	if _, found := c.CraftingState.Attestation.Materials[key]; found {
-		c.logger.Info().Str("key", key).Str("value", value).Msg("material already set, overriding it")
+		c.Logger.Info().Str("key", key).Str("value", value).Msg("material already set, overriding it")
 	}
 
 	// 3 - Craft resulting material
@@ -518,7 +526,7 @@ func (c *Crafter) AddMaterialContactFreeAutomatic(ctx context.Context, attestati
 			return kind, nil
 		}
 
-		c.logger.Debug().Err(err).Str("kind", kind.String()).Msg("failed to add material")
+		c.Logger.Debug().Err(err).Str("kind", kind.String()).Msg("failed to add material")
 
 		// Handle base error for upload and craft errors except the opening file error
 		var policyError *policies.PolicyError
@@ -534,7 +542,7 @@ func (c *Crafter) AddMaterialContactFreeAutomatic(ctx context.Context, attestati
 // addMaterials adds the incoming material m to the crafting state
 func (c *Crafter) addMaterial(ctx context.Context, m *schemaapi.CraftingSchema_Material, attestationID, value string, casBackend *casclient.CASBackend, runtimeAnnotations map[string]string) error {
 	// 3- Craft resulting material
-	mt, err := materials.Craft(context.Background(), m, value, casBackend, c.ociRegistryAuth, c.logger)
+	mt, err := materials.Craft(context.Background(), m, value, casBackend, c.ociRegistryAuth, c.Logger)
 	if err != nil {
 		return err
 	}
@@ -551,7 +559,7 @@ func (c *Crafter) addMaterial(ctx context.Context, m *schemaapi.CraftingSchema_M
 			mt.Annotations[kr] = vr
 		} else {
 			// NOTE: we do not allow overriding values that come from the contract
-			c.logger.Info().Str("key", m.Name).Str("annotation", kr).Msg("annotation can't be changed, skipping")
+			c.Logger.Info().Str("key", m.Name).Str("annotation", kr).Msg("annotation can't be changed, skipping")
 		}
 	}
 
@@ -573,13 +581,13 @@ func (c *Crafter) addMaterial(ctx context.Context, m *schemaapi.CraftingSchema_M
 	}
 
 	// Validate policies
-	pv := policies.NewPolicyVerifier(c.CraftingState.InputSchema, c.logger)
+	pv := policies.NewPolicyVerifier(c.CraftingState.InputSchema, c.Logger)
 	policyResults, err := pv.VerifyMaterial(ctx, mt, value)
 	if err != nil {
 		return fmt.Errorf("error applying policies to material: %w", err)
 	}
 	// log policy violations
-	policies.LogPolicyViolations(policyResults, c.logger)
+	policies.LogPolicyViolations(policyResults, c.Logger)
 	// store policy results
 	c.CraftingState.Attestation.PolicyEvaluations = append(c.CraftingState.Attestation.PolicyEvaluations, policyResults...)
 
@@ -594,7 +602,7 @@ func (c *Crafter) addMaterial(ctx context.Context, m *schemaapi.CraftingSchema_M
 		return fmt.Errorf("failed to persist crafting state: %w", err)
 	}
 
-	c.logger.Debug().Str("key", m.Name).Msg("added to state")
+	c.Logger.Debug().Str("key", m.Name).Msg("added to state")
 	return nil
 }
 

internal/attestation/crafter/statemanager/filesystem/filesystem.go

@@ -21,7 +21,7 @@ import (
 	"io"
 	"os"
 
-	v1 "github.com/chainloop-dev/chainloop/internal/attestation/crafter/api/attestation/v1"
+	"github.com/chainloop-dev/chainloop/internal/attestation/crafter"
 	"github.com/chainloop-dev/chainloop/internal/attestation/crafter/statemanager"
 	"google.golang.org/protobuf/encoding/protojson"
 )
@@ -55,7 +55,7 @@ func (l *Filesystem) Initialized(_ context.Context, _ string) (bool, error) {
 	return false, nil
 }
 
-func (l *Filesystem) Write(_ context.Context, _ string, state *v1.CraftingState) error {
+func (l *Filesystem) Write(_ context.Context, _ string, state *crafter.VersionedCraftingState) error {
 	if state == nil {
 		return fmt.Errorf("state cannot be nil")
 	}
@@ -81,7 +81,7 @@ func (l *Filesystem) Write(_ context.Context, _ string, state *v1.CraftingState)
 	return nil
 }
 
-func (l *Filesystem) Read(_ context.Context, _ string, state *v1.CraftingState) error {
+func (l *Filesystem) Read(_ context.Context, _ string, state *crafter.VersionedCraftingState) error {
 	if state == nil {
 		return fmt.Errorf("state cannot be nil")
 	}

internal/attestation/crafter/statemanager/filesystem/filesystem_test.go

@@ -21,6 +21,7 @@ import (
 	"os"
 	"testing"
 
+	"github.com/chainloop-dev/chainloop/internal/attestation/crafter"
 	v1 "github.com/chainloop-dev/chainloop/internal/attestation/crafter/api/attestation/v1"
 	"github.com/chainloop-dev/chainloop/internal/attestation/crafter/statemanager"
 	"github.com/stretchr/testify/require"
@@ -61,7 +62,7 @@ func (s *testSuite) TestNew() {
 func (s *testSuite) TestWrite() {
 	testCases := []struct {
 		name    string
-		state   *v1.CraftingState
+		state   *crafter.VersionedCraftingState
 		wantErr bool
 	}{
 		{
@@ -87,7 +88,7 @@ func (s *testSuite) TestWrite() {
 			}
 
 			s.NoError(err)
-			got := &v1.CraftingState{}
+			got := &crafter.VersionedCraftingState{CraftingState: &v1.CraftingState{}}
 			err = sm.Read(context.Background(), "", got)
 			s.NoError(err)
 			s.Equal(tc.state, got)
@@ -107,7 +108,7 @@ func (s *testSuite) TestRead() {
 	s.T().Run("no state found in path return NotFound error", func(t *testing.T) {
 		sm, err := New(s.statePath)
 		require.NoError(t, err)
-		err = sm.Read(context.Background(), "", &v1.CraftingState{})
+		err = sm.Read(context.Background(), "", &crafter.VersionedCraftingState{})
 		s.Error(err)
 		want := &statemanager.ErrNotFound{}
 		s.ErrorAs(err, &want)
@@ -116,7 +117,7 @@ func (s *testSuite) TestRead() {
 	s.T().Run("we can read the state", func(t *testing.T) {
 		sm, err := New("testdata/state.json")
 		require.NoError(t, err)
-		got := &v1.CraftingState{}
+		got := &crafter.VersionedCraftingState{CraftingState: &v1.CraftingState{}}
 		err = sm.Read(context.Background(), "", got)
 		require.NoError(s.T(), err)
 
@@ -173,14 +174,14 @@ func (s *testSuite) TestInitialized() {
 type testSuite struct {
 	suite.Suite
 	statePath    string
-	exampleState *v1.CraftingState
+	exampleState *crafter.VersionedCraftingState
 }
 
 func (s *testSuite) SetupTest() {
 	s.statePath = fmt.Sprintf("%s/attestation.json", s.T().TempDir())
-	s.exampleState = &v1.CraftingState{DryRun: true, Attestation: &v1.Attestation{
+	s.exampleState = &crafter.VersionedCraftingState{CraftingState: &v1.CraftingState{DryRun: true, Attestation: &v1.Attestation{
 		Annotations: map[string]string{"foo": "bar"},
-	}}
+	}}}
 }
 
 func TestSuite(t *testing.T) {