feat: improved account handling for db transient errors - closes #85

.github/workflows/sonar.yml

@@ -16,7 +16,7 @@ jobs:
     steps:
       - name: Get PR details
         id: pr
-        uses: actions/github-script@v7
+        uses: actions/github-script@v9
         with:
           script: |
             const pr = await github.rest.pulls.get({

go.mod

@@ -16,7 +16,7 @@ require (
 	dario.cat/mergo v1.0.2 // indirect
 	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/ProtonMail/go-crypto v1.4.1 // indirect
-	github.com/cloudflare/circl v1.6.3 // indirect
+	github.com/cloudflare/circl v1.6.4 // indirect
 	github.com/cyphar/filepath-securejoin v0.7.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/emirpasic/gods v1.18.1 // indirect

go.sum

@@ -9,8 +9,8 @@ github.com/anmitsu/go-shlex v0.0.0-20200514113438-38f4b401e2be h1:9AeTilPcZAjCFI
 github.com/anmitsu/go-shlex v0.0.0-20200514113438-38f4b401e2be/go.mod h1:ySMOLuWl6zY27l47sB3qLNK6tF2fkHG55UZxx8oIVo4=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
-github.com/cloudflare/circl v1.6.3 h1:9GPOhQGF9MCYUeXyMYlqTR6a5gTrgR/fBLXvUgtVcg8=
-github.com/cloudflare/circl v1.6.3/go.mod h1:2eXP6Qfat4O/Yhh8BznvKnJ+uzEoTQ6jVKJRn81BiS4=
+github.com/cloudflare/circl v1.6.4 h1:pOXuDTCEYyzydgUpQ0CQz3LsinKjiSk6nNP5Lt5K64U=
+github.com/cloudflare/circl v1.6.4/go.mod h1:YxarevkLlbaHuWsxG6vmYNWBEsSp4pnp7j+4VljMavY=
 github.com/cyphar/filepath-securejoin v0.7.0 h1:s0Y3ITPy6sQn5xt54DuYvTF8hu134ooYLUb58DX/HjE=
 github.com/cyphar/filepath-securejoin v0.7.0/go.mod h1:ymLGms/u3BYaviIiuKFnUx8EkQEZeK6cInNoAPJA3o4=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

src/core/handler.go

@@ -1,6 +1,7 @@
 package core
 
 import (
+	"errors"
 	"fmt"
 	"time"
 
@@ -69,49 +70,32 @@ func (c *CoreHandler) StartAccountHandler(accountId string, gitService IGitServi
 	utils.LogInfo("[%s] Starting account handler", accountId)
 
 	for {
-		// Refresh account settings
-		account, _ := c.accountRepository.FetchByAccountId(accountId)
-
-		if account == nil {
-			// Terminate the goroutine (account was deleted)
-			utils.LogInfo("[%s] Account was deleted, terminating account handler", accountId)
+		account, shouldStop := c.fetchActiveAccount(accountId)
+		if shouldStop {
 			return
 		}
+		if account == nil {
+			continue
+		}
 
 		// Wait for account to be active
 		if !account.Settings.Active {
 			c.updateDomainStatus(*account, model.StatusPending, "Account was deactivated",
 				utils.LogLevelInfo)
-			time.Sleep(time.Duration(c.waitingTime))
+			c.waitForNextAttempt()
 			continue
 		}
 
 		// Refresh account repository settings
 		gitService.UpdateRepositorySettings(account.Repository, account.Token, account.Branch, account.Path)
 
-		// Fetch repository data
-		repositoryData, err := c.getRepositoryData(gitService, account)
-
-		if err != nil {
-			c.updateDomainStatus(*account, model.StatusError, "Failed to fetch repository data - "+err.Error(),
-				utils.LogLevelError)
-			time.Sleep(time.Duration(c.waitingTime))
+		repositoryData, shouldContinue := c.fetchRepositoryData(account, gitService)
+		if shouldContinue {
 			continue
 		}
 
-		if !utils.IsJsonValid(repositoryData.Content, &model.Snapshot{}) {
-			c.updateDomainStatus(*account, model.StatusError, "Invalid JSON content", utils.LogLevelError)
-			time.Sleep(time.Duration(c.waitingTime))
-			continue
-		}
-
-		// Fetch snapshot version from API
-		snapshotVersionPayload, err := c.apiService.FetchSnapshotVersion(account.Domain.ID, account.Environment)
-
-		if err != nil {
-			c.updateDomainStatus(*account, model.StatusError, "Failed to fetch snapshot version - "+err.Error(),
-				utils.LogLevelError)
-			time.Sleep(time.Duration(c.waitingTime))
+		snapshotVersionPayload, shouldContinue := c.fetchSnapshotVersion(account)
+		if shouldContinue {
 			continue
 		}
 
@@ -120,12 +104,70 @@ func (c *CoreHandler) StartAccountHandler(accountId string, gitService IGitServi
 			c.syncUp(*account, repositoryData, gitService)
 		}
 
-		// Wait for the next cycle
-		timeWindow, unitWindow := utils.GetTimeWindow(account.Settings.Window)
-		time.Sleep(time.Duration(timeWindow) * unitWindow)
+		c.waitForAccountWindow(account)
 	}
 }
 
+func (c *CoreHandler) fetchActiveAccount(accountId string) (*model.Account, bool) {
+	account, err := c.accountRepository.FetchByAccountId(accountId)
+	if err == nil && account != nil {
+		return account, false
+	}
+
+	if errors.Is(err, repository.ErrAccountNotFound) {
+		utils.LogInfo("[%s] Account was deleted, terminating account handler", accountId)
+		return nil, true
+	}
+
+	if err != nil {
+		utils.LogError("[%s] Failed to refresh account settings - %s", accountId, err.Error())
+	} else {
+		utils.LogError("[%s] Failed to refresh account settings - empty account payload", accountId)
+	}
+
+	c.waitForNextAttempt()
+	return nil, false
+}
+
+func (c *CoreHandler) fetchRepositoryData(account *model.Account, gitService IGitService) (*model.RepositoryData, bool) {
+	repositoryData, err := c.getRepositoryData(gitService, account)
+	if err != nil {
+		c.updateDomainStatus(*account, model.StatusError, "Failed to fetch repository data - "+err.Error(),
+			utils.LogLevelError)
+		c.waitForNextAttempt()
+		return nil, true
+	}
+
+	if !utils.IsJsonValid(repositoryData.Content, &model.Snapshot{}) {
+		c.updateDomainStatus(*account, model.StatusError, "Invalid JSON content", utils.LogLevelError)
+		c.waitForNextAttempt()
+		return nil, true
+	}
+
+	return repositoryData, false
+}
+
+func (c *CoreHandler) fetchSnapshotVersion(account *model.Account) (string, bool) {
+	snapshotVersionPayload, err := c.apiService.FetchSnapshotVersion(account.Domain.ID, account.Environment)
+	if err != nil {
+		c.updateDomainStatus(*account, model.StatusError, "Failed to fetch snapshot version - "+err.Error(),
+			utils.LogLevelError)
+		c.waitForNextAttempt()
+		return "", true
+	}
+
+	return snapshotVersionPayload, false
+}
+
+func (c *CoreHandler) waitForNextAttempt() {
+	time.Sleep(c.waitingTime)
+}
+
+func (c *CoreHandler) waitForAccountWindow(account *model.Account) {
+	timeWindow, unitWindow := utils.GetTimeWindow(account.Settings.Window)
+	time.Sleep(time.Duration(timeWindow) * unitWindow)
+}
+
 func (c *CoreHandler) getRepositoryData(gitService IGitService, account *model.Account) (*model.RepositoryData, error) {
 	repositoryData, err := gitService.GetRepositoryData(account.Environment)
 

src/core/handler_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/switcherapi/switcher-gitops/src/model"
+	"github.com/switcherapi/switcher-gitops/src/repository"
 )
 
 func TestInitCoreHandlerGoroutine(t *testing.T) {
@@ -346,6 +347,37 @@ func TestAccountHandlerSyncAPI(t *testing.T) {
 }
 
 func TestAccountHandlerNotSync(t *testing.T) {
+	t.Run("Should retry when account fetch returns an empty account payload", func(t *testing.T) {
+		// Given
+		flakyRepository := &EmptyFetchAccountRepository{
+			AccountRepository:   repository.NewAccountRepositoryMongo(mongoDb),
+			emptyFetchByIDCount: 1,
+		}
+		fakeApiService := NewFakeApiService()
+
+		coreHandler = NewCoreHandler(flakyRepository, fakeApiService, NewComparatorService())
+		coreHandler.waitingTime = 10 * time.Millisecond
+
+		account := givenAccount()
+		account.Domain.ID = "123-empty-payload"
+		accountCreated, _ := flakyRepository.Create(&account)
+
+		// Test
+		accountFromFetch, shouldStop := coreHandler.fetchActiveAccount(accountCreated.ID.Hex())
+
+		// Assert
+		assert.Nil(t, accountFromFetch)
+		assert.False(t, shouldStop)
+		assert.Equal(t, 1, flakyRepository.fetchByIDCalls)
+
+		accountFromFetch, shouldStop = coreHandler.fetchActiveAccount(accountCreated.ID.Hex())
+		assert.NotNil(t, accountFromFetch)
+		assert.False(t, shouldStop)
+		assert.Equal(t, 2, flakyRepository.fetchByIDCalls)
+
+		tearDown()
+	})
+
 	t.Run("Should not sync when account is not active", func(t *testing.T) {
 		// Given
 		fakeGitService := NewFakeGitService()
@@ -447,6 +479,38 @@ func TestAccountHandlerNotSync(t *testing.T) {
 		tearDown()
 	})
 
+	t.Run("Should retry when fetch snapshot version returns a transient error", func(t *testing.T) {
+		// Given
+		fakeGitService := NewFakeGitService()
+		flakyApiService := &RetrySnapshotVersionApiService{
+			FakeApiService:                NewFakeApiService(),
+			failFetchSnapshotVersionCount: 1,
+		}
+
+		coreHandler = NewCoreHandler(coreHandler.accountRepository, flakyApiService, NewComparatorService())
+		coreHandler.waitingTime = 10 * time.Millisecond
+
+		account := givenAccount()
+		account.Domain.ID = "123-retry-fetch-snapshot-version"
+		accountCreated, _ := coreHandler.accountRepository.Create(&account)
+
+		// Test
+		go coreHandler.StartAccountHandler(accountCreated.ID.Hex(), fakeGitService)
+
+		time.Sleep(200 * time.Millisecond)
+
+		// Assert
+		accountFromDb, _ := coreHandler.accountRepository.FetchByDomainIdEnvironment(accountCreated.Domain.ID, accountCreated.Environment)
+		assert.GreaterOrEqual(t, flakyApiService.fetchSnapshotVersionCalls, 2)
+		assert.Equal(t, model.StatusSynced, accountFromDb.Domain.Status)
+		assert.Contains(t, accountFromDb.Domain.Message, model.MessageSynced)
+		assert.Equal(t, "123", accountFromDb.Domain.LastCommit)
+		assert.Equal(t, 1, accountFromDb.Domain.Version)
+		assert.NotEqual(t, "", accountFromDb.Domain.LastDate)
+
+		tearDown()
+	})
+
 	t.Run("Should not sync after account is deleted", func(t *testing.T) {
 		// Given
 		fakeGitService := NewFakeGitService()
@@ -472,6 +536,39 @@ func TestAccountHandlerNotSync(t *testing.T) {
 		tearDown()
 	})
 
+	t.Run("Should retry when account fetch returns a transient error", func(t *testing.T) {
+		// Given
+		fakeGitService := NewFakeGitService()
+		fakeApiService := NewFakeApiService()
+		flakyRepository := &RetryFetchAccountRepository{
+			AccountRepository:  repository.NewAccountRepositoryMongo(mongoDb),
+			failFetchByIDCount: 1,
+		}
+
+		coreHandler = NewCoreHandler(flakyRepository, fakeApiService, NewComparatorService())
+		coreHandler.waitingTime = 10 * time.Millisecond
+
+		account := givenAccount()
+		account.Domain.ID = "123-fetch-retry"
+		accountCreated, _ := flakyRepository.Create(&account)
+
+		// Test
+		go coreHandler.StartAccountHandler(accountCreated.ID.Hex(), fakeGitService)
+
+		time.Sleep(200 * time.Millisecond)
+
+		// Assert
+		accountFromDb, _ := flakyRepository.FetchByDomainIdEnvironment(accountCreated.Domain.ID, accountCreated.Environment)
+		assert.GreaterOrEqual(t, flakyRepository.fetchByIDCalls, 2)
+		assert.Equal(t, model.StatusSynced, accountFromDb.Domain.Status)
+		assert.Contains(t, accountFromDb.Domain.Message, model.MessageSynced)
+		assert.Equal(t, "123", accountFromDb.Domain.LastCommit)
+		assert.Equal(t, 1, accountFromDb.Domain.Version)
+		assert.NotEqual(t, "", accountFromDb.Domain.LastDate)
+
+		tearDown()
+	})
+
 	t.Run("Should not sync when API returns an error", func(t *testing.T) {
 		// Given
 		fakeGitService := NewFakeGitService()
@@ -628,6 +725,51 @@ func tearDown() {
 
 // Fakes
 
+type RetryFetchAccountRepository struct {
+	repository.AccountRepository
+	failFetchByIDCount int
+	fetchByIDCalls     int
+}
+
+func (r *RetryFetchAccountRepository) FetchByAccountId(accountId string) (*model.Account, error) {
+	r.fetchByIDCalls++
+	if r.fetchByIDCalls <= r.failFetchByIDCount {
+		return nil, errors.New("temporary db outage")
+	}
+
+	return r.AccountRepository.FetchByAccountId(accountId)
+}
+
+type EmptyFetchAccountRepository struct {
+	repository.AccountRepository
+	emptyFetchByIDCount int
+	fetchByIDCalls      int
+}
+
+func (r *EmptyFetchAccountRepository) FetchByAccountId(accountId string) (*model.Account, error) {
+	r.fetchByIDCalls++
+	if r.fetchByIDCalls <= r.emptyFetchByIDCount {
+		return nil, nil
+	}
+
+	return r.AccountRepository.FetchByAccountId(accountId)
+}
+
+type RetrySnapshotVersionApiService struct {
+	*FakeApiService
+	failFetchSnapshotVersionCount int
+	fetchSnapshotVersionCalls     int
+}
+
+func (f *RetrySnapshotVersionApiService) FetchSnapshotVersion(domainId, environment string) (string, error) {
+	f.fetchSnapshotVersionCalls++
+	if f.fetchSnapshotVersionCalls <= f.failFetchSnapshotVersionCount {
+		return "", errors.New("temporary snapshot version outage")
+	}
+
+	return f.FakeApiService.FetchSnapshotVersion(domainId, environment)
+}
+
 type FakeGitService struct {
 	existingData     model.RepositoryData
 	newData          model.RepositoryData