opensource/coder
3
0
Fork 0
mirror of https://github.com/coder/coder.git synced 2026-09-24 15:04:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore: clean up dbpurge after the chatd refactor (#26344)

Browse Source 
Addresses
https://github.com/coder/coder/pull/26109#discussion_r3379072397 and
https://github.com/coder/coder/pull/26109#discussion_r3379093655.
This commit is contained in:
Hugo Dutka
2026-06-16 14:01:31 +02:00
committed by GitHub
parent f08bb652b4
commit 62288782fc
3 changed files with 31 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cli/server.go
+1 -1
View File
@@ -1158,7 +1158,7 @@ func (r *RootCmd) Server(newAPI func(context.Context, *coderd.Options) (*coderd.
defer shutdownConns()
// Ensures that old database entries are cleaned up over time!
purger := dbpurge.New(ctx, logger.Named("dbpurge"), options.Database, options.DeploymentValues, options.PrometheusRegistry, &coderAPI.Auditor)
purger := dbpurge.New(ctx, logger.Named("dbpurge"), options.Database, options.DeploymentValues, options.PrometheusRegistry)
defer purger.Close()
// Updates workspace usage
coderd/database/dbpurge/dbpurge.go
+5 -8
View File
@@ -4,14 +4,12 @@ import (
"context"
"errors"
"io"
"sync/atomic"
"time"
"github.com/prometheus/client_golang/prometheus"
"golang.org/x/xerrors"
"cdr.dev/slog/v3"
"github.com/coder/coder/v2/coderd/audit"
"github.com/coder/coder/v2/coderd/database"
"github.com/coder/coder/v2/coderd/database/dbauthz"
"github.com/coder/coder/v2/coderd/database/dbtime"
@@ -56,10 +54,7 @@ func WithClock(clk quartz.Clock) Option {
// New creates a new periodically purging database instance.
// Callers must Close the returned instance.
//
// The auditor pointer is accepted for compatibility with other background
// services. Dbpurge does not emit audit logs directly.
func New(ctx context.Context, logger slog.Logger, db database.Store, vals *codersdk.DeploymentValues, reg prometheus.Registerer, _ *atomic.Pointer[audit.Auditor], opts ...Option) io.Closer {
func New(ctx context.Context, logger slog.Logger, db database.Store, vals *codersdk.DeploymentValues, reg prometheus.Registerer, opts ...Option) io.Closer {
closed := make(chan struct{})
ctx, cancelFunc := context.WithCancel(ctx)
@@ -139,11 +134,13 @@ func (i *instance) purgeTick(ctx context.Context, db database.Store, start time.
// the tx so the failed iteration is operator-visible via metric and
// logs.
chatRetentionDays, chatRetentionErr := db.GetChatRetentionDays(ctx)
purgeChats := chatRetentionErr == nil
if chatRetentionErr != nil {
i.logger.Error(ctx, "failed to read chat retention config: skipping chat purge this tick", slog.Error(chatRetentionErr))
}
chatDebugRetentionDays, chatDebugRetentionErr := db.GetChatDebugRetentionDays(ctx, codersdk.DefaultChatDebugRetentionDays)
purgeChatDebugRuns := chatDebugRetentionErr == nil
if chatDebugRetentionErr != nil {
i.logger.Error(ctx, "failed to read chat debug retention config: skipping chat debug purge this tick", slog.Error(chatDebugRetentionErr))
}
@@ -255,13 +252,13 @@ func (i *instance) purgeTick(ctx context.Context, db database.Store, start time.
}
var purgedChats, purgedChatFiles, purgedChatDebugRuns int64
if chatRetentionErr == nil {
if purgeChats {
purgedChats, purgedChatFiles, err = i.purgeChatsInTx(ctx, tx, start, chatRetentionDays)
if err != nil {
return xerrors.Errorf("failed to purge chats: %w", err)
}
}
if chatDebugRetentionErr == nil && chatDebugRetentionDays > 0 {
if purgeChatDebugRuns && chatDebugRetentionDays > 0 {
deleteChatDebugRunsBefore := start.Add(-time.Duration(chatDebugRetentionDays) * 24 * time.Hour)
// updated_at is the retention clock, so the window starts after
// the run stops being written to. There is intentionally no
coderd/database/dbpurge/dbpurge_test.go
+25 -36
View File
@@ -8,7 +8,6 @@ import (
"encoding/json"
"fmt"
"slices"
"sync/atomic"
"testing"
"time"
@@ -23,7 +22,6 @@ import (
"cdr.dev/slog/v3"
"cdr.dev/slog/v3/sloggers/slogtest"
"github.com/coder/coder/v2/coderd/audit"
"github.com/coder/coder/v2/coderd/coderdtest/promhelp"
"github.com/coder/coder/v2/coderd/database"
"github.com/coder/coder/v2/coderd/database/dbgen"
@@ -59,7 +57,7 @@ func TestPurge(t *testing.T) {
mDB.EXPECT().GetChatRetentionDays(gomock.Any()).Return(int32(0), nil).AnyTimes()
mDB.EXPECT().GetChatDebugRetentionDays(gomock.Any(), codersdk.DefaultChatDebugRetentionDays).Return(int32(0), nil).AnyTimes()
mDB.EXPECT().InTx(gomock.Any(), database.DefaultTXOptions().WithID("db_purge")).Return(nil).Times(2)
purger := dbpurge.New(context.Background(), testutil.Logger(t), mDB, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
purger := dbpurge.New(context.Background(), testutil.Logger(t), mDB, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
<-done // wait for doTick() to run.
require.NoError(t, purger.Close())
}
@@ -93,7 +91,7 @@ func TestMetrics(t *testing.T) {
Retention: codersdk.RetentionConfig{
APIKeys: serpent.Duration(7 * 24 * time.Hour), // 7 days retention
},
}, reg, nopAuditorPtr(t), dbpurge.WithClock(clk))
}, reg, dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -170,7 +168,7 @@ func TestMetrics(t *testing.T) {
logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, mDB, &codersdk.DeploymentValues{}, reg, nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, mDB, &codersdk.DeploymentValues{}, reg, dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -206,7 +204,7 @@ func TestMetrics(t *testing.T) {
logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, mDB, &codersdk.DeploymentValues{}, reg, nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, mDB, &codersdk.DeploymentValues{}, reg, dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -259,7 +257,7 @@ func TestMetrics(t *testing.T) {
logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, mDB, &codersdk.DeploymentValues{}, reg, nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, mDB, &codersdk.DeploymentValues{}, reg, dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -310,7 +308,7 @@ func TestMetrics(t *testing.T) {
logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, mDB, &codersdk.DeploymentValues{}, reg, nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, mDB, &codersdk.DeploymentValues{}, reg, dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -401,7 +399,7 @@ func TestDeleteOldWorkspaceAgentStats(t *testing.T) {
})
// when
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
// then
@@ -426,7 +424,7 @@ func TestDeleteOldWorkspaceAgentStats(t *testing.T) {
// Start a new purger to immediately trigger delete after rollup.
_ = closer.Close()
closer = dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
closer = dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
// then
@@ -521,7 +519,7 @@ func TestDeleteOldWorkspaceAgentLogs(t *testing.T) {
Retention: codersdk.RetentionConfig{
WorkspaceAgentLogs: serpent.Duration(7 * 24 * time.Hour),
},
}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
<-done // doTick() has now run.
@@ -736,7 +734,7 @@ func TestDeleteOldWorkspaceAgentLogsRetention(t *testing.T) {
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{
Retention: tc.retentionConfig,
}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -827,7 +825,7 @@ func TestDeleteOldProvisionerDaemons(t *testing.T) {
require.NoError(t, err)
// when
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
// then
@@ -931,7 +929,7 @@ func TestDeleteOldAuditLogConnectionEvents(t *testing.T) {
// Run the purge
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
// Wait for tick
testutil.TryReceive(ctx, t, done)
@@ -1094,7 +1092,7 @@ func TestDeleteOldTelemetryHeartbeats(t *testing.T) {
require.NoError(t, err)
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
<-done // doTick() has now run.
@@ -1213,7 +1211,7 @@ func TestDeleteOldConnectionLogs(t *testing.T) {
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{
Retention: tc.retentionConfig,
}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -1469,7 +1467,7 @@ func TestDeleteOldAIBridgeRecords(t *testing.T) {
Retention: serpent.Duration(tc.retention),
},
},
}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -1556,7 +1554,7 @@ func TestDeleteOldAuditLogs(t *testing.T) {
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{
Retention: tc.retentionConfig,
}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -1646,7 +1644,7 @@ func TestDeleteOldAuditLogs(t *testing.T) {
Retention: codersdk.RetentionConfig{
AuditLogs: serpent.Duration(retentionPeriod),
},
}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -1766,7 +1764,7 @@ func TestDeleteExpiredAPIKeys(t *testing.T) {
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{
Retention: tc.retentionConfig,
}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -1801,15 +1799,6 @@ func ptr[T any](v T) *T {
return &v
}
// nopAuditorPtr returns an atomic pointer to a nop auditor for tests.
func nopAuditorPtr(t *testing.T) *atomic.Pointer[audit.Auditor] {
t.Helper()
nop := audit.NewNop()
var p atomic.Pointer[audit.Auditor]
p.Store(&nop)
return &p
}
//nolint:paralleltest // It uses LockIDDBPurge.
func TestPurgeChatDebugRuns(t *testing.T) {
now := time.Date(2025, 6, 15, 12, 0, 0, 0, time.UTC)
@@ -1921,7 +1910,7 @@ func TestPurgeChatDebugRuns(t *testing.T) {
unfinishedOldRun := createDebugRunWithStep(ctx, t, db, chat.ID, now.Add(-9*24*time.Hour), false)
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, reg, nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, reg, dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -1959,7 +1948,7 @@ func TestPurgeChatDebugRuns(t *testing.T) {
oldRun := createDebugRunWithStep(ctx, t, db, chat.ID, now.Add(-90*24*time.Hour), true)
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -1985,7 +1974,7 @@ func TestPurgeChatDebugRuns(t *testing.T) {
run := createDebugRunWithStep(ctx, t, db, oldArchivedChat.ID, now, true)
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -2091,7 +2080,7 @@ func TestDeleteOldChatFiles(t *testing.T) {
oldFileID := createChatFile(ctx, t, db, rawDB, deps.user.ID, deps.org.ID, now.Add(-31*24*time.Hour))
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -2133,7 +2122,7 @@ func TestDeleteOldChatFiles(t *testing.T) {
activeChat := createChat(ctx, t, db, rawDB, deps.user.ID, deps.org.ID, deps.modelConfig.ID, false, now)
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -2190,7 +2179,7 @@ func TestDeleteOldChatFiles(t *testing.T) {
fileBoundary := createChatFile(ctx, t, db, rawDB, deps.user.ID, deps.org.ID, now.Add(-30*24*time.Hour).Add(time.Hour))
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)
@@ -2270,7 +2259,7 @@ func TestDeleteOldChatFiles(t *testing.T) {
require.NoError(t, err)
done := awaitDoTick(ctx, t, clk)
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), nopAuditorPtr(t), dbpurge.WithClock(clk))
closer := dbpurge.New(ctx, logger, db, &codersdk.DeploymentValues{}, prometheus.NewRegistry(), dbpurge.WithClock(clk))
defer closer.Close()
testutil.TryReceive(ctx, t, done)