Geyser Timelock worker fixes to detect unlock state of unitialized accounts

Author: jeffyanta

pkg/code/async/geyser/backup.go

@@ -36,13 +36,13 @@ func (p *service) backupTimelockStateWorker(serviceCtx context.Context, state ti
 		log.Debug("worker stopped")
 	}()
 
-	delay := 0 * time.Second // Initially no delay, so we can run right after a deploy
+	start := time.Now()
 	cursor := query.EmptyCursor
-	oldestRecordTs := time.Now()
+	delay := 0 * time.Second // Initially no delay, so we can run right after a deploy
 	for {
 		select {
 		case <-time.After(delay):
-			start := time.Now()
+			batchStart := time.Now()
 
 			func() {
 				nr := serviceCtx.Value(metrics.NewRelicContextKey).(*newrelic.Application)
@@ -59,14 +59,14 @@ func (p *service) backupTimelockStateWorker(serviceCtx context.Context, state ti
 				)
 				if err == timelock.ErrTimelockNotFound {
 					p.metricStatusLock.Lock()
-					copiedTs := oldestRecordTs
-					if p.oldestTimelockRecord == nil || p.oldestTimelockRecord.After(copiedTs) {
-						p.oldestTimelockRecord = &copiedTs
+					duration := time.Since(start)
+					if p.backupTimelockStateWorkerDuration == nil || *p.backupTimelockStateWorkerDuration < duration {
+						p.backupTimelockStateWorkerDuration = &duration
 					}
 					p.metricStatusLock.Unlock()
 
+					start = time.Now()
 					cursor = query.EmptyCursor
-					oldestRecordTs = time.Now()
 					return
 				} else if err != nil {
 					log.WithError(err).Warn("failed to get timelock records")
@@ -77,10 +77,6 @@ func (p *service) backupTimelockStateWorker(serviceCtx context.Context, state ti
 				for _, timelockRecord := range timelockRecords {
 					wg.Add(1)
 
-					if timelockRecord.LastUpdatedAt.Before(oldestRecordTs) {
-						oldestRecordTs = timelockRecord.LastUpdatedAt
-					}
-
 					go func(timelockRecord *timelock.Record) {
 						defer wg.Done()
 
@@ -98,7 +94,7 @@ func (p *service) backupTimelockStateWorker(serviceCtx context.Context, state ti
 				cursor = query.ToCursor(timelockRecords[len(timelockRecords)-1].Id)
 			}()
 
-			delay = interval - time.Since(start)
+			delay = interval - time.Since(batchStart)
 		case <-serviceCtx.Done():
 			return serviceCtx.Err()
 		}

pkg/code/async/geyser/config.go

@@ -23,10 +23,10 @@ const (
 	defaultProgramUpdateQueueSize       = 1_000_000
 
 	BackupTimelockWorkerIntervalConfigEnvName = envConfigPrefix + "BACKUP_TIMELOCK_WORKER_INTERVAL"
-	defaultBackupTimelockWorkerInterval       = 1 * time.Minute
+	defaultBackupTimelockWorkerInterval       = time.Second
 
 	BackupExternalDepositWorkerIntervalConfigEnvName = envConfigPrefix + "BACKUP_EXTERNAL_DEPOSIT_WORKER_INTERVAL"
-	defaultBackupExternalDepositWorkerInterval       = 15 * time.Second
+	defaultBackupExternalDepositWorkerInterval       = time.Second
 )
 
 type conf struct {

pkg/code/async/geyser/metrics.go

@@ -109,10 +109,10 @@ func (p *service) recordBackupWorkerStatusPollingEvent(ctx context.Context) {
 		"worker_type": timelockStateWorkerName,
 		"is_active":   p.backupTimelockStateWorkerStatus,
 	}
-	if p.oldestTimelockRecord != nil {
-		oldestRecordAgeSeconds := time.Since(*p.oldestTimelockRecord) / time.Second
-		timelockMetrics["oldest_record_age_s"] = int(oldestRecordAgeSeconds)
-		p.oldestTimelockRecord = nil
+	if p.backupTimelockStateWorkerDuration != nil {
+		inSeconds := *p.backupTimelockStateWorkerDuration / time.Second
+		timelockMetrics["duration_s"] = int(inSeconds)
+		p.backupTimelockStateWorkerDuration = nil
 	}
 	metrics.RecordEvent(ctx, backupWorkerStatusEventName, timelockMetrics)
 

pkg/code/async/geyser/service.go

@@ -41,8 +41,8 @@ type service struct {
 	slotUpdateSubscriptionStatus bool
 	highestObservedFinalizedSlot uint64
 
-	oldestTimelockRecord            *time.Time
-	backupTimelockStateWorkerStatus bool
+	backupTimelockStateWorkerDuration *time.Duration
+	backupTimelockStateWorkerStatus   bool
 
 	backupExternalDepositWorkerStatus bool
 }
@@ -69,12 +69,12 @@ func (p *service) Start(ctx context.Context, _ time.Duration) error {
 			p.log.WithError(err).Warn("timelock backup worker terminated unexpectedly")
 		}
 	}()
-	/*go func() {
+	go func() {
 		err := p.backupTimelockStateWorker(ctx, timelock_token.StateUnknown, p.conf.backupTimelockWorkerInterval.Get(ctx))
 		if err != nil && err != context.Canceled {
 			p.log.WithError(err).Warn("timelock backup worker terminated unexpectedly")
 		}
-	}()*/
+	}()
 
 	go func() {
 		err := p.backupExternalDepositWorker(ctx, p.conf.backupExternalDepositWorkerInterval.Get(ctx))

pkg/code/async/geyser/timelock.go

@@ -27,9 +27,13 @@ func updateTimelockAccountRecord(ctx context.Context, data code_data.Provider, t
 		unlockAt := uint64(unlockState.UnlockAt)
 		timelockRecord.UnlockAt = &unlockAt
 	}
+
+	if timelockRecord.VaultState == timelock_token.StateUnknown {
+		return nil
+	}
+
 	timelockRecord.Block = slot
 	timelockRecord.LastUpdatedAt = time.Now()
-
 	return data.SaveTimelock(ctx, timelockRecord)
 }