@@ -2,11 +2,10 @@ package dbpurge
22
33import (
44"context"
5- "errors"
65"io"
76"time"
87
9- "golang.org/x/sync/errgroup "
8+ "golang.org/x/xerrors "
109
1110"cdr.dev/slog"
1211
@@ -35,22 +34,37 @@ func New(ctx context.Context, logger slog.Logger, db database.Store) io.Closer {
3534doTick := func () {
3635defer ticker .Reset (delay )
3736
38- var eg errgroup.Group
39- eg .Go (func ()error {
40- return db .DeleteOldWorkspaceAgentLogs (ctx )
41- })
42- eg .Go (func ()error {
43- return db .DeleteOldWorkspaceAgentStats (ctx )
44- })
45- eg .Go (func ()error {
46- return db .DeleteOldProvisionerDaemons (ctx )
47- })
48- err := eg .Wait ()
49- if err != nil {
50- if errors .Is (err ,context .Canceled ) {
51- return
37+ start := time .Now ()
38+ // Start a transaction to grab advisory lock, we don't want to run
39+ // multiple purges at the same time (multiple replicas).
40+ if err := db .InTx (func (tx database.Store )error {
41+ // Acquire a lock to ensure that only one instance of the
42+ // purge is running at a time.
43+ ok ,err := tx .TryAcquireLock (ctx ,database .LockIDDBPurge )
44+ if err != nil {
45+ return err
46+ }
47+ if ! ok {
48+ logger .Debug (ctx ,"unable to acquire lock for purging old database entries, skipping" )
49+ return nil
50+ }
51+
52+ if err := tx .DeleteOldWorkspaceAgentLogs (ctx );err != nil {
53+ return xerrors .Errorf ("failed to delete old workspace agent logs: %w" ,err )
5254}
55+ if err := tx .DeleteOldWorkspaceAgentStats (ctx );err != nil {
56+ return xerrors .Errorf ("failed to delete old workspace agent stats: %w" ,err )
57+ }
58+ if err := tx .DeleteOldProvisionerDaemons (ctx );err != nil {
59+ return xerrors .Errorf ("failed to delete old provisioner daemons: %w" ,err )
60+ }
61+
62+ logger .Info (ctx ,"purged old database entries" ,slog .F ("duration" ,time .Since (start )))
63+
64+ return nil
65+ },nil );err != nil {
5366logger .Error (ctx ,"failed to purge old database entries" ,slog .Error (err ))
67+ return
5468}
5569}
5670