Merge pull request #934 from laverya/async-state-update

sync state update

Author: laverya

pkg/lifecycle/daemon/routes_navcycle_completestep.go

@@ -12,6 +12,7 @@ import (
 	"github.com/replicatedhq/ship/pkg/api"
 	"github.com/replicatedhq/ship/pkg/lifecycle/daemon/daemontypes"
 	"github.com/replicatedhq/ship/pkg/lifecycle/daemon/statusonly"
+	"github.com/replicatedhq/ship/pkg/state"
 	"github.com/replicatedhq/ship/pkg/util/warnings"
 )
 
@@ -67,14 +68,9 @@ func (d *NavcycleRoutes) handleAsync(errChan chan error, debug log.Logger, step
 		return
 	}
 
-	state, err := d.StateManager.TryLoad()
-	if err != nil {
-		level.Error(d.Logger).Log("event", "state.load.fail", "err", err)
-		return
-	}
-
-	newState := state.Versioned().WithCompletedStep(step)
-	err = d.StateManager.Save(newState)
+	_, err := d.StateManager.StateUpdate(func(currentState state.VersionedState) (state.VersionedState, error) {
+		return currentState.WithCompletedStep(step), nil
+	})
 	if err != nil {
 		level.Error(d.Logger).Log("event", "state.save.fail", "err", err, "step.id", stepID)
 		return

pkg/lifecycle/daemon/routes_navcycle_completestep_test.go

@@ -17,9 +17,9 @@ import (
 	"github.com/replicatedhq/ship/pkg/templates"
 	"github.com/replicatedhq/ship/pkg/test-mocks/lifecycle"
 	planner2 "github.com/replicatedhq/ship/pkg/test-mocks/planner"
-	"github.com/replicatedhq/ship/pkg/test-mocks/state"
 	"github.com/replicatedhq/ship/pkg/testing/logger"
 	"github.com/replicatedhq/ship/pkg/testing/matchers"
+	"github.com/spf13/afero"
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/require"
 )
@@ -333,16 +333,18 @@ func TestV2CompleteStep(t *testing.T) {
 					},
 				},
 			}
-			mc := gomock.NewController(t)
-			fakeState := state.NewMockManager(mc)
+
 			testLogger := &logger.TestLogger{T: t}
+			fs := afero.Afero{Fs: afero.NewMemMapFs()}
+			realState := state2.NewManager(testLogger, fs, viper.New())
+			mc := gomock.NewController(t)
 			messenger := lifecycle.NewMockMessenger(mc)
 			renderer := lifecycle.NewMockRenderer(mc)
 			mockPlanner := planner2.NewMockPlanner(mc)
 			v2 := &NavcycleRoutes{
-				BuilderBuilder: templates.NewBuilderBuilder(testLogger, viper.New(), fakeState),
+				BuilderBuilder: templates.NewBuilderBuilder(testLogger, viper.New(), realState),
 				Logger:         testLogger,
-				StateManager:   fakeState,
+				StateManager:   realState,
 				Messenger:      messenger,
 				Renderer:       renderer,
 				Planner:        mockPlanner,
@@ -352,14 +354,6 @@ func TestV2CompleteStep(t *testing.T) {
 				StepProgress: &daemontypes.ProgressMap{},
 			}
 
-			fakeState.EXPECT().TryLoad().Return(state2.VersionedState{
-				V1: &state2.V1{
-					Lifecycle: &state2.Lifeycle{
-						StepsCompleted: make(map[string]interface{}),
-					},
-				},
-			}, nil).AnyTimes()
-
 			if test.OnExecute != nil {
 				v2.StepExecutor = test.OnExecute
 			}
@@ -379,10 +373,6 @@ func TestV2CompleteStep(t *testing.T) {
 
 				// send request
 				for _, testCase := range test.POSTS {
-					if testCase.ExpectState != nil && testCase.ExpectState.Test != nil {
-						fakeState.EXPECT().Save(testCase.ExpectState).Return(nil)
-					}
-
 					resp, err := http.Post(fmt.Sprintf("%s%s", addr, testCase.POST), "application/json", strings.NewReader(""))
 					req.NoError(err)
 					req.Equal(testCase.ExpectStatus, resp.StatusCode)
@@ -398,6 +388,15 @@ func TestV2CompleteStep(t *testing.T) {
 						bodyForDebug = []byte(err.Error())
 					}
 					req.Empty(diff, "\nexpect: %s\nactual: %s\ndiff: %s", bodyForDebug, string(bytes), strings.Join(diff, "\n"))
+
+					if testCase.ExpectState != nil && testCase.ExpectState.Test != nil {
+						// there is almost certainly a better solution than this
+						time.Sleep(time.Duration(time.Millisecond * 500))
+
+						loadState, err := realState.TryLoad()
+						req.NoError(err)
+						req.True(testCase.ExpectState.Test(loadState), testCase.ExpectState.Describe)
+					}
 				}
 			}()
 		})

pkg/lifecycle/render/config/daemonresolver_test.go

@@ -8,10 +8,6 @@ import (
 
 	"github.com/mitchellh/cli"
 	"github.com/replicatedhq/libyaml"
-	"github.com/spf13/afero"
-	"github.com/spf13/viper"
-	"github.com/stretchr/testify/require"
-
 	"github.com/replicatedhq/ship/pkg/api"
 	"github.com/replicatedhq/ship/pkg/lifecycle/daemon"
 	"github.com/replicatedhq/ship/pkg/lifecycle/daemon/headless"
@@ -21,6 +17,9 @@ import (
 	templates "github.com/replicatedhq/ship/pkg/templates"
 	"github.com/replicatedhq/ship/pkg/testing/logger"
 	"github.com/replicatedhq/ship/pkg/ui"
+	"github.com/spf13/afero"
+	"github.com/spf13/viper"
+	"github.com/stretchr/testify/require"
 )
 
 type daemonResolverTestCase struct {

pkg/specs/persist.go

@@ -7,7 +7,6 @@ import (
 	"strings"
 
 	"github.com/pkg/errors"
-
 	"github.com/replicatedhq/ship/pkg/state"
 )
 

pkg/specs/replicatedapp/graphql.go

@@ -14,7 +14,6 @@ import (
 	"github.com/pkg/errors"
 	"github.com/replicatedhq/ship/pkg/api"
 	"github.com/replicatedhq/ship/pkg/state"
-
 	"github.com/spf13/viper"
 )
 

pkg/specs/stategetter/client_test.go

@@ -6,12 +6,10 @@ import (
 	"path/filepath"
 	"testing"
 
-	"github.com/stretchr/testify/require"
-
 	"github.com/replicatedhq/ship/pkg/state"
 	"github.com/replicatedhq/ship/pkg/testing/logger"
-
 	"github.com/spf13/afero"
+	"github.com/stretchr/testify/require"
 )
 
 func TestStateGetter_GetFiles(t *testing.T) {

pkg/state/manager.go

@@ -53,11 +53,12 @@ var _ Manager = &MManager{}
 
 // MManager is the saved output of a plan run to load on future runs
 type MManager struct {
-	Logger  log.Logger
-	FS      afero.Afero
-	V       *viper.Viper
-	patcher patch.Patcher
-	mut     sync.Mutex
+	Logger         log.Logger
+	FS             afero.Afero
+	V              *viper.Viper
+	patcher        patch.Patcher
+	stateUpdateMut sync.Mutex
+	StateRWMut     sync.RWMutex
 }
 
 func (m *MManager) Save(v VersionedState) error {
@@ -87,8 +88,8 @@ type Update func(VersionedState) (VersionedState, error)
 
 // applies the provided updater to the current state. Returns the new state and err
 func (m *MManager) StateUpdate(updater Update) (State, error) {
-	m.mut.Lock()
-	defer m.mut.Unlock()
+	m.stateUpdateMut.Lock()
+	defer m.stateUpdateMut.Unlock()
 
 	currentState, err := m.TryLoad()
 	if err != nil {
@@ -266,6 +267,8 @@ func (m *MManager) SerializeUpstreamContents(contents *UpstreamContents) error {
 
 // TryLoad will attempt to load a state file from disk, if present
 func (m *MManager) TryLoad() (State, error) {
+	m.StateRWMut.RLock()
+	defer m.StateRWMut.RUnlock()
 	stateFrom := m.V.GetString("state-from")
 	if stateFrom == "" {
 		stateFrom = "file"
@@ -426,6 +429,8 @@ func (m *MManager) RemoveStateFile() error {
 }
 
 func (m *MManager) serializeAndWriteState(state VersionedState) error {
+	m.StateRWMut.Lock()
+	defer m.StateRWMut.Unlock()
 	debug := level.Debug(log.With(m.Logger, "method", "serializeAndWriteState"))
 	state = state.migrateDeprecatedFields()
 

pkg/state/manager_test.go

@@ -11,7 +11,6 @@ import (
 	"github.com/replicatedhq/ship/pkg/constants"
 	"github.com/replicatedhq/ship/pkg/testing/logger"
 	"github.com/replicatedhq/ship/pkg/util"
-
 	"github.com/spf13/afero"
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/require"