Delete old replicas on a CRON

kylecarbs · kylecarbs · commit ae0aa5f226bd · 2022-10-15T20:55:11.000Z
diff --git a/coderd/database/databasefake/databasefake.go b/coderd/database/databasefake/databasefake.go
@@ -3238,15 +3238,3 @@ func (q *fakeQuerier) GetReplicasUpdatedAfter(_ context.Context, updatedAt time.
 	}
 	return replicas, nil
 }
-
-func (q *fakeQuerier) GetReplicaByID(_ context.Context, id uuid.UUID) (database.Replica, error) {
-	q.mutex.RLock()
-	defer q.mutex.RUnlock()
-
-	for _, replica := range q.replicas {
-		if replica.ID == id {
-			return replica, nil
-		}
-	}
-	return database.Replica{}, sql.ErrNoRows
-}
diff --git a/coderd/database/querier.go b/coderd/database/querier.go
diff --git a/coderd/database/queries.sql.go b/coderd/database/queries.sql.go
diff --git a/coderd/database/queries/replicas.sql b/coderd/database/queries/replicas.sql
@@ -1,9 +1,6 @@
 -- name: GetReplicasUpdatedAfter :many
 SELECT * FROM replicas WHERE updated_at > $1 AND stopped_at IS NULL;
 
--- name: GetReplicaByID :one
-SELECT * FROM replicas WHERE id = $1;
-
 -- name: InsertReplica :one
 INSERT INTO replicas (
     id,
diff --git a/enterprise/replicasync/replicasync.go b/enterprise/replicasync/replicasync.go
@@ -26,11 +26,12 @@ var (
 )
 
 type Options struct {
-	UpdateInterval time.Duration
-	PeerTimeout    time.Duration
-	RelayAddress   string
-	RegionID       int32
-	TLSConfig      *tls.Config
+	CleanupInterval time.Duration
+	UpdateInterval  time.Duration
+	PeerTimeout     time.Duration
+	RelayAddress    string
+	RegionID        int32
+	TLSConfig       *tls.Config
 }
 
 // New registers the replica with the database and periodically updates to ensure
@@ -45,6 +46,11 @@ func New(ctx context.Context, logger slog.Logger, db database.Store, pubsub data
 	if options.UpdateInterval == 0 {
 		options.UpdateInterval = 5 * time.Second
 	}
+	if options.CleanupInterval == 0 {
+		// The cleanup interval can be quite long, because it's
+		// primary purpose is to clean up dead replicas.
+		options.CleanupInterval = 30 * time.Minute
+	}
 	hostname, err := os.Hostname()
 	if err != nil {
 		return nil, xerrors.Errorf("get hostname: %w", err)
@@ -123,16 +129,31 @@ type Manager struct {
 	callback func()
 }
 
+// updateInterval is used to determine a replicas state.
+// If the replica was updated > the time, it's considered healthy.
+// If the replica was updated < the time, it's considered stale.
+func (m *Manager) updateInterval() time.Time {
+	return database.Now().Add(-3 * m.options.UpdateInterval)
+}
+
 // loop runs the replica update sequence on an update interval.
 func (m *Manager) loop(ctx context.Context) {
 	defer m.closeWait.Done()
-	ticker := time.NewTicker(m.options.UpdateInterval)
-	defer ticker.Stop()
+	updateTicker := time.NewTicker(m.options.UpdateInterval)
+	defer updateTicker.Stop()
+	deleteTicker := time.NewTicker(m.options.CleanupInterval)
+	defer deleteTicker.Stop()
 	for {
 		select {
 		case <-ctx.Done():
 			return
-		case <-ticker.C:
+		case <-deleteTicker.C:
+			err := m.db.DeleteReplicasUpdatedBefore(ctx, m.updateInterval())
+			if err != nil {
+				m.logger.Warn(ctx, "delete old replicas", slog.Error(err))
+			}
+			continue
+		case <-updateTicker.C:
 		}
 		err := m.syncReplicas(ctx)
 		if err != nil && !errors.Is(err, context.Canceled) {
@@ -204,7 +225,7 @@ func (m *Manager) syncReplicas(ctx context.Context) error {
 	defer m.closeWait.Done()
 	// Expect replicas to update once every three times the interval...
 	// If they don't, assume death!
-	replicas, err := m.db.GetReplicasUpdatedAfter(ctx, database.Now().Add(-3*m.options.UpdateInterval))
+	replicas, err := m.db.GetReplicasUpdatedAfter(ctx, m.updateInterval())
 	if err != nil {
 		return xerrors.Errorf("get replicas: %w", err)
 	}
diff --git a/enterprise/replicasync/replicasync_test.go b/enterprise/replicasync/replicasync_test.go
@@ -178,6 +178,24 @@ func TestReplica(t *testing.T) {
 		}, testutil.WaitShort, testutil.IntervalFast)
 		_ = server.Close()
 	})
+	t.Run("DeletesOld", func(t *testing.T) {
+		t.Parallel()
+		db, pubsub := dbtestutil.NewDB(t)
+		_, err := db.InsertReplica(context.Background(), database.InsertReplicaParams{
+			ID:        uuid.New(),
+			UpdatedAt: database.Now().Add(-time.Hour),
+		})
+		require.NoError(t, err)
+		server, err := replicasync.New(context.Background(), slogtest.Make(t, nil), db, pubsub, &replicasync.Options{
+			RelayAddress:    "google.com",
+			CleanupInterval: time.Millisecond,
+		})
+		require.NoError(t, err)
+		defer server.Close()
+		require.Eventually(t, func() bool {
+			return len(server.Regional()) == 0
+		}, testutil.WaitShort, testutil.IntervalFast)
+	})
 	t.Run("TwentyConcurrent", func(t *testing.T) {
 		// Ensures that twenty concurrent replicas can spawn and all
 		// discover each other in parallel!
