Added scheduled sync implementation

Author: akclace

internal/server/sync.go

@@ -162,6 +162,16 @@ func (s *Server) runSyncJobs() error {
 			continue
 		}
 
+		if !entry.Status.LastExecutionTime.IsZero() && entry.Status.LastExecutionTime.Add(time.Duration(entry.Metadata.ScheduleFrequency)*time.Minute).After(time.Now()) {
+			s.Trace().Msgf("Sync job %s not ready to run", entry.Id)
+			continue
+		}
+
+		if entry.Status.FailureCount >= s.config.System.MaxSyncFailureCount {
+			s.Trace().Msgf("Sync job %s has failed too many times, skipping", entry.Id)
+			continue
+		}
+
 		_, _, err = s.runSyncJob(ctx, types.Transaction{}, entry, true, repoCache) // each sync runs in its own transaction
 		if err != nil {
 			s.Error().Err(err).Msgf("Error running sync job %s", entry.Id)
@@ -172,7 +182,8 @@ func (s *Server) runSyncJobs() error {
 	return nil
 }
 
-func (s *Server) runSyncJob(ctx context.Context, inputTx types.Transaction, entry *types.SyncEntry, checkCommitHash bool, repoCache *RepoCache) (*types.SyncJobStatus, []types.AppPathDomain, error) {
+func (s *Server) runSyncJob(ctx context.Context, inputTx types.Transaction, entry *types.SyncEntry,
+	checkCommitHash bool, repoCache *RepoCache) (*types.SyncJobStatus, []types.AppPathDomain, error) {
 	var tx types.Transaction
 	var err error
 	if inputTx.Tx == nil {
@@ -186,6 +197,7 @@ func (s *Server) runSyncJob(ctx context.Context, inputTx types.Transaction, entr
 		// No rollback here if transaction is passed in
 	}
 
+	s.Debug().Msgf("Running sync job %s", entry.Id)
 	if repoCache == nil {
 		// Create a new repo cache if not passed in
 		repoCache, err = NewRepoCache(s)
@@ -213,8 +225,10 @@ func (s *Server) runSyncJob(ctx context.Context, inputTx types.Transaction, entr
 		status.Error = applyErr.Error()
 		applyInfo = &types.AppApplyResponse{}
 		applyInfo.FilteredApps = lastRunApps
+		status.FailureCount++
 	} else {
 		status.CommitId = applyInfo.CommitId
+		status.FailureCount = 0
 	}
 
 	status.ApplyResponse = *applyInfo
@@ -270,8 +284,8 @@ func (s *Server) runSyncJob(ctx context.Context, inputTx types.Transaction, entr
 			status.ApplyResponse.PromoteResults = promoteResults
 		}
 	}
-	status.ApplyResponse = *applyInfo
 
+	status.ApplyResponse = *applyInfo
 	err = s.db.UpdateSyncStatus(ctx, tx, entry.Id, &status)
 	if err != nil {
 		return nil, nil, err

internal/system/clace.default.toml

@@ -64,6 +64,7 @@ default_domain = "localhost"        # default domain for apps
 root_serve_list_apps = "auto"       # "auto" means serve list_apps app for default domain, "disable" means don't server for any domain,
 enable_compression = false          # enable compression for HTTP responses
 default_schedule_mins = 5           # default sync schedule interval in minutes
+max_sync_failure_count = 5          # max number of sync failures before sync is marked as disabled
 # any other value means server for specified domain
 
 http_event_retention_days = 90      # number of days to retain http events

internal/types/types.go

@@ -218,8 +218,9 @@ type SystemConfig struct {
 	EnableCompression         bool     `toml:"enable_compression"`
 	HttpEventRetentionDays    int      `toml:"http_event_retention_days"`
 	NonHttpEventRetentionDays int      `toml:"non_http_event_retention_days"`
-	AllowedEnv                []string `toml:"allowed_env"`           // List of environment variables that are allowed to be used in the node config
-	DefaultScheduleMins       int      `toml:"default_schedule_mins"` // Default schedule time in minutes for scheduled sync
+	AllowedEnv                []string `toml:"allowed_env"`            // List of environment variables that are allowed to be used in the node config
+	DefaultScheduleMins       int      `toml:"default_schedule_mins"`  // Default schedule time in minutes for scheduled sync
+	MaxSyncFailureCount       int      `toml:"max_sync_failure_count"` // Max failure count for sync jobs
 }
 
 // GitAuth is a github auth config entry
@@ -604,6 +605,7 @@ type SyncMetadata struct {
 }
 
 type SyncJobStatus struct {
+	FailureCount      int              `json:"failure_count"`       // the number of times the sync job has failed recently
 	LastExecutionTime time.Time        `json:"last_execution_time"` // the last time the sync job was executed
 	Error             string           `json:"error"`               // the error message if the sync job failed
 	CommitId          string           `json:"commit_id"`           // the commit id of the sync job