detached c318835

fix(backup): propagate contexts and harden scheduler behavior

commit
c31883553d1bde404f3696703ce30dbcb976d6d3
author
brice · · 2026-02-07 16:20 UTC
parent
1ac3076
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
9 files · +180 -30
Unified
UnifiedSplit
M internal/adapters/in/cli/backup.go +4 -5
@@ -1,7 +1,6 @@
11 package cli
22
33 import (
4- "context"
54 "fmt"
65 "os"
76 "text/tabwriter"
@@ -44,7 +43,7 @@ func newBackupListCmd() *cobra.Command {
4443 domainName = args[0]
4544 }
4645
47- jobs, err := client.ListBackups(context.Background(), domainName)
46+ jobs, err := client.ListBackups(cmd.Context(), domainName)
4847 if err != nil {
4948 return fmt.Errorf("failed to list backups: %w", err)
5049 }
@@ -85,7 +84,7 @@ func newBackupRunCmd() *cobra.Command {
8584 return fmt.Errorf("backup commands require a configured remote target")
8685 }
8786
88- result, err := client.RunBackup(context.Background(), args[0], dbName)
87+ result, err := client.RunBackup(cmd.Context(), args[0], dbName)
8988 if err != nil {
9089 return fmt.Errorf("failed to run backup: %w", err)
9190 }
@@ -123,7 +122,7 @@ func newBackupDetectCmd() *cobra.Command {
123122 return fmt.Errorf("backup commands require a configured remote target")
124123 }
125124
126- dbs, err := client.DetectDatabases(context.Background(), args[0])
125+ dbs, err := client.DetectDatabases(cmd.Context(), args[0])
127126 if err != nil {
128127 return fmt.Errorf("failed to detect databases: %w", err)
129128 }
@@ -161,7 +160,7 @@ func newBackupStatusCmd() *cobra.Command {
161160 return fmt.Errorf("backup commands require a configured remote target")
162161 }
163162
164- jobs, err := client.BackupStatus(context.Background())
163+ jobs, err := client.BackupStatus(cmd.Context())
165164 if err != nil {
166165 return fmt.Errorf("failed to get backup status: %w", err)
167166 }
@@ -1,7 +1,6 @@
1 package cli1 package cli
2 2
3 import (3 import (
4- "context"
5 "fmt"4 "fmt"
6 "os"5 "os"
7 "text/tabwriter"6 "text/tabwriter"
@@ -44,7 +43,7 @@ func newBackupListCmd() *cobra.Command {
44 domainName = args[0]43 domainName = args[0]
45 }44 }
46 45
47- jobs, err := client.ListBackups(context.Background(), domainName)46+ jobs, err := client.ListBackups(cmd.Context(), domainName)
48 if err != nil {47 if err != nil {
49 return fmt.Errorf("failed to list backups: %w", err)48 return fmt.Errorf("failed to list backups: %w", err)
50 }49 }
@@ -85,7 +84,7 @@ func newBackupRunCmd() *cobra.Command {
85 return fmt.Errorf("backup commands require a configured remote target")84 return fmt.Errorf("backup commands require a configured remote target")
86 }85 }
87 86
88- result, err := client.RunBackup(context.Background(), args[0], dbName)87+ result, err := client.RunBackup(cmd.Context(), args[0], dbName)
89 if err != nil {88 if err != nil {
90 return fmt.Errorf("failed to run backup: %w", err)89 return fmt.Errorf("failed to run backup: %w", err)
91 }90 }
@@ -123,7 +122,7 @@ func newBackupDetectCmd() *cobra.Command {
123 return fmt.Errorf("backup commands require a configured remote target")122 return fmt.Errorf("backup commands require a configured remote target")
124 }123 }
125 124
126- dbs, err := client.DetectDatabases(context.Background(), args[0])125+ dbs, err := client.DetectDatabases(cmd.Context(), args[0])
127 if err != nil {126 if err != nil {
128 return fmt.Errorf("failed to detect databases: %w", err)127 return fmt.Errorf("failed to detect databases: %w", err)
129 }128 }
@@ -161,7 +160,7 @@ func newBackupStatusCmd() *cobra.Command {
161 return fmt.Errorf("backup commands require a configured remote target")160 return fmt.Errorf("backup commands require a configured remote target")
162 }161 }
163 162
164- jobs, err := client.BackupStatus(context.Background())163+ jobs, err := client.BackupStatus(cmd.Context())
165 if err != nil {164 if err != nil {
166 return fmt.Errorf("failed to get backup status: %w", err)165 return fmt.Errorf("failed to get backup status: %w", err)
167 }166 }
M internal/adapters/in/http/admin/handler.go +2 -0
@@ -770,6 +770,8 @@ func (h *Handler) handleBackups(w http.ResponseWriter, r *http.Request, path str
770770 return
771771 }
772772
773+ path = strings.TrimSuffix(path, "/")
774+
773775 if path == "/backups" || path == "/backups/status" {
774776 h.handleBackupsStatus(w, r)
775777 return
@@ -770,6 +770,8 @@ func (h *Handler) handleBackups(w http.ResponseWriter, r *http.Request, path str
770 return770 return
771 }771 }
772 772
773+ path = strings.TrimSuffix(path, "/")
774+
773 if path == "/backups" || path == "/backups/status" {775 if path == "/backups" || path == "/backups/status" {
774 h.handleBackupsStatus(w, r)776 h.handleBackupsStatus(w, r)
775 return777 return
M internal/adapters/out/docker/runtime.go +32 -1
@@ -7,12 +7,14 @@ import (
77 "context"
88 "encoding/base64"
99 "encoding/json"
10+ "errors"
1011 "fmt"
1112 "io"
1213 "path/filepath"
1314 "sort"
1415 "strconv"
1516 "strings"
17+ "sync"
1618 "time"
1719
1820 "github.com/bnema/zerowrap"
@@ -35,6 +37,35 @@ type Runtime struct {
3537 client *client.Client
3638 }
3739
40+type pipeReadCloser struct {
41+ pr *io.PipeReader
42+ pw *io.PipeWriter
43+ original io.Closer
44+ once sync.Once
45+}
46+
47+func (p *pipeReadCloser) Read(b []byte) (int, error) {
48+ return p.pr.Read(b)
49+}
50+
51+func (p *pipeReadCloser) Close() error {
52+ var closeErr error
53+ p.once.Do(func() {
54+ if err := p.pw.CloseWithError(errors.New("reader closed")); err != nil && !errors.Is(err, io.ErrClosedPipe) {
55+ closeErr = err
56+ }
57+ if err := p.pr.Close(); err != nil && closeErr == nil {
58+ closeErr = err
59+ }
60+ if p.original != nil {
61+ if err := p.original.Close(); err != nil && closeErr == nil {
62+ closeErr = err
63+ }
64+ }
65+ })
66+ return closeErr
67+}
68+
3869 // NewRuntime creates a new Docker runtime instance.
3970 func NewRuntime() (*Runtime, error) {
4071 cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
@@ -1238,7 +1269,7 @@ func extractFileFromTar(reader io.ReadCloser, targetPath string) (io.ReadCloser,
12381269 }
12391270 _ = pw.Close()
12401271 }()
1241- return pr, nil
1272+ return &pipeReadCloser{pr: pr, pw: pw, original: reader}, nil
12421273 }
12431274 }
12441275 }
@@ -7,12 +7,14 @@ import (
7 "context"7 "context"
8 "encoding/base64"8 "encoding/base64"
9 "encoding/json"9 "encoding/json"
10+ "errors"
10 "fmt"11 "fmt"
11 "io"12 "io"
12 "path/filepath"13 "path/filepath"
13 "sort"14 "sort"
14 "strconv"15 "strconv"
15 "strings"16 "strings"
17+ "sync"
16 "time"18 "time"
17 19
18 "github.com/bnema/zerowrap"20 "github.com/bnema/zerowrap"
@@ -35,6 +37,35 @@ type Runtime struct {
35 client *client.Client37 client *client.Client
36 }38 }
37 39
40+type pipeReadCloser struct {
41+ pr *io.PipeReader
42+ pw *io.PipeWriter
43+ original io.Closer
44+ once sync.Once
45+}
46+
47+func (p *pipeReadCloser) Read(b []byte) (int, error) {
48+ return p.pr.Read(b)
49+}
50+
51+func (p *pipeReadCloser) Close() error {
52+ var closeErr error
53+ p.once.Do(func() {
54+ if err := p.pw.CloseWithError(errors.New("reader closed")); err != nil && !errors.Is(err, io.ErrClosedPipe) {
55+ closeErr = err
56+ }
57+ if err := p.pr.Close(); err != nil && closeErr == nil {
58+ closeErr = err
59+ }
60+ if p.original != nil {
61+ if err := p.original.Close(); err != nil && closeErr == nil {
62+ closeErr = err
63+ }
64+ }
65+ })
66+ return closeErr
67+}
68+
38 // NewRuntime creates a new Docker runtime instance.69 // NewRuntime creates a new Docker runtime instance.
39 func NewRuntime() (*Runtime, error) {70 func NewRuntime() (*Runtime, error) {
40 cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())71 cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
@@ -1238,7 +1269,7 @@ func extractFileFromTar(reader io.ReadCloser, targetPath string) (io.ReadCloser,
1238 }1269 }
1239 _ = pw.Close()1270 _ = pw.Close()
1240 }()1271 }()
1241- return pr, nil1272+ return &pipeReadCloser{pr: pr, pw: pw, original: reader}, nil
1242 }1273 }
1243 }1274 }
1244 }1275 }
M internal/app/run_backup_test.go +28 -11
@@ -56,16 +56,33 @@ func TestValidateBackupRetentionRejectsNegativeValues(t *testing.T) {
5656 }
5757
5858 func TestValidateBackupRetentionAcceptsZeroAndPositiveValues(t *testing.T) {
59- var cfg Config
60- cfg.Backups.Retention.Hourly = 1
61- cfg.Backups.Retention.Daily = 7
62- cfg.Backups.Retention.Weekly = 4
63- cfg.Backups.Retention.Monthly = 12
59+ t.Run("positive values", func(t *testing.T) {
60+ var cfg Config
61+ cfg.Backups.Retention.Hourly = 1
62+ cfg.Backups.Retention.Daily = 7
63+ cfg.Backups.Retention.Weekly = 4
64+ cfg.Backups.Retention.Monthly = 12
6465
65- retention, err := validateBackupRetention(cfg)
66- require.NoError(t, err)
67- assert.Equal(t, 1, retention.Hourly)
68- assert.Equal(t, 7, retention.Daily)
69- assert.Equal(t, 4, retention.Weekly)
70- assert.Equal(t, 12, retention.Monthly)
66+ retention, err := validateBackupRetention(cfg)
67+ require.NoError(t, err)
68+ assert.Equal(t, 1, retention.Hourly)
69+ assert.Equal(t, 7, retention.Daily)
70+ assert.Equal(t, 4, retention.Weekly)
71+ assert.Equal(t, 12, retention.Monthly)
72+ })
73+
74+ t.Run("zero values", func(t *testing.T) {
75+ var cfg Config
76+ cfg.Backups.Retention.Hourly = 0
77+ cfg.Backups.Retention.Daily = 0
78+ cfg.Backups.Retention.Weekly = 0
79+ cfg.Backups.Retention.Monthly = 0
80+
81+ retention, err := validateBackupRetention(cfg)
82+ require.NoError(t, err)
83+ assert.Equal(t, 0, retention.Hourly)
84+ assert.Equal(t, 0, retention.Daily)
85+ assert.Equal(t, 0, retention.Weekly)
86+ assert.Equal(t, 0, retention.Monthly)
87+ })
7188 }
@@ -56,16 +56,33 @@ func TestValidateBackupRetentionRejectsNegativeValues(t *testing.T) {
56 }56 }
57 57
58 func TestValidateBackupRetentionAcceptsZeroAndPositiveValues(t *testing.T) {58 func TestValidateBackupRetentionAcceptsZeroAndPositiveValues(t *testing.T) {
59- var cfg Config59+ t.Run("positive values", func(t *testing.T) {
60- cfg.Backups.Retention.Hourly = 160+ var cfg Config
61- cfg.Backups.Retention.Daily = 761+ cfg.Backups.Retention.Hourly = 1
62- cfg.Backups.Retention.Weekly = 462+ cfg.Backups.Retention.Daily = 7
63- cfg.Backups.Retention.Monthly = 1263+ cfg.Backups.Retention.Weekly = 4
64+ cfg.Backups.Retention.Monthly = 12
64 65
65- retention, err := validateBackupRetention(cfg)66+ retention, err := validateBackupRetention(cfg)
66- require.NoError(t, err)67+ require.NoError(t, err)
67- assert.Equal(t, 1, retention.Hourly)68+ assert.Equal(t, 1, retention.Hourly)
68- assert.Equal(t, 7, retention.Daily)69+ assert.Equal(t, 7, retention.Daily)
69- assert.Equal(t, 4, retention.Weekly)70+ assert.Equal(t, 4, retention.Weekly)
70- assert.Equal(t, 12, retention.Monthly)71+ assert.Equal(t, 12, retention.Monthly)
72+ })
73+
74+ t.Run("zero values", func(t *testing.T) {
75+ var cfg Config
76+ cfg.Backups.Retention.Hourly = 0
77+ cfg.Backups.Retention.Daily = 0
78+ cfg.Backups.Retention.Weekly = 0
79+ cfg.Backups.Retention.Monthly = 0
80+
81+ retention, err := validateBackupRetention(cfg)
82+ require.NoError(t, err)
83+ assert.Equal(t, 0, retention.Hourly)
84+ assert.Equal(t, 0, retention.Daily)
85+ assert.Equal(t, 0, retention.Weekly)
86+ assert.Equal(t, 0, retention.Monthly)
87+ })
71 }88 }
M internal/usecase/backup/integration_test.go +21 -6
@@ -45,8 +45,14 @@ func runPostgresBackupFlow(t *testing.T, ctx context.Context, runtime *docker.Ru
4545 pullCtx, cancelPull := context.WithTimeout(ctx, 5*time.Minute)
4646 defer cancelPull()
4747 require.NoError(t, runtime.PullImage(pullCtx, image))
48- require.NoError(t, runtime.CreateNetwork(ctx, networkName, map[string]string{"driver": "bridge"}))
49- t.Cleanup(func() { _ = runtime.RemoveNetwork(context.Background(), networkName) })
48+ netCtx, cancelNet := context.WithTimeout(ctx, 2*time.Minute)
49+ defer cancelNet()
50+ require.NoError(t, runtime.CreateNetwork(netCtx, networkName, map[string]string{"driver": "bridge"}))
51+ t.Cleanup(func() {
52+ cleanupCtx, cancelCleanup := context.WithTimeout(context.Background(), 20*time.Second)
53+ defer cancelCleanup()
54+ _ = runtime.RemoveNetwork(cleanupCtx, networkName)
55+ })
5056
5157 containerCfg := &domain.ContainerConfig{
5258 Image: image,
@@ -66,13 +72,22 @@ func runPostgresBackupFlow(t *testing.T, ctx context.Context, runtime *docker.Ru
6672 },
6773 }
6874
69- container, err := runtime.CreateContainer(ctx, containerCfg)
75+ contCtx, cancelCont := context.WithTimeout(ctx, 2*time.Minute)
76+ defer cancelCont()
77+ container, err := runtime.CreateContainer(contCtx, containerCfg)
7078 require.NoError(t, err)
7179 t.Cleanup(func() {
72- _ = runtime.StopContainer(context.Background(), container.ID)
73- _ = runtime.RemoveContainer(context.Background(), container.ID, true)
80+ stopCtx, cancelStop := context.WithTimeout(context.Background(), 20*time.Second)
81+ _ = runtime.StopContainer(stopCtx, container.ID)
82+ cancelStop()
83+
84+ removeCtx, cancelRemove := context.WithTimeout(context.Background(), 20*time.Second)
85+ _ = runtime.RemoveContainer(removeCtx, container.ID, true)
86+ cancelRemove()
7487 })
75- require.NoError(t, runtime.StartContainer(ctx, container.ID))
88+ startCtx, cancelStart := context.WithTimeout(ctx, 2*time.Minute)
89+ defer cancelStart()
90+ require.NoError(t, runtime.StartContainer(startCtx, container.ID))
7691
7792 require.NoError(t, waitForPostgresReady(ctx, runtime, container.ID, 60*time.Second))
7893 require.NoError(t, seedPostgresData(ctx, runtime, container.ID))
@@ -45,8 +45,14 @@ func runPostgresBackupFlow(t *testing.T, ctx context.Context, runtime *docker.Ru
45 pullCtx, cancelPull := context.WithTimeout(ctx, 5*time.Minute)45 pullCtx, cancelPull := context.WithTimeout(ctx, 5*time.Minute)
46 defer cancelPull()46 defer cancelPull()
47 require.NoError(t, runtime.PullImage(pullCtx, image))47 require.NoError(t, runtime.PullImage(pullCtx, image))
48- require.NoError(t, runtime.CreateNetwork(ctx, networkName, map[string]string{"driver": "bridge"}))48+ netCtx, cancelNet := context.WithTimeout(ctx, 2*time.Minute)
49- t.Cleanup(func() { _ = runtime.RemoveNetwork(context.Background(), networkName) })49+ defer cancelNet()
50+ require.NoError(t, runtime.CreateNetwork(netCtx, networkName, map[string]string{"driver": "bridge"}))
51+ t.Cleanup(func() {
52+ cleanupCtx, cancelCleanup := context.WithTimeout(context.Background(), 20*time.Second)
53+ defer cancelCleanup()
54+ _ = runtime.RemoveNetwork(cleanupCtx, networkName)
55+ })
50 56
51 containerCfg := &domain.ContainerConfig{57 containerCfg := &domain.ContainerConfig{
52 Image: image,58 Image: image,
@@ -66,13 +72,22 @@ func runPostgresBackupFlow(t *testing.T, ctx context.Context, runtime *docker.Ru
66 },72 },
67 }73 }
68 74
69- container, err := runtime.CreateContainer(ctx, containerCfg)75+ contCtx, cancelCont := context.WithTimeout(ctx, 2*time.Minute)
76+ defer cancelCont()
77+ container, err := runtime.CreateContainer(contCtx, containerCfg)
70 require.NoError(t, err)78 require.NoError(t, err)
71 t.Cleanup(func() {79 t.Cleanup(func() {
72- _ = runtime.StopContainer(context.Background(), container.ID)80+ stopCtx, cancelStop := context.WithTimeout(context.Background(), 20*time.Second)
73- _ = runtime.RemoveContainer(context.Background(), container.ID, true)81+ _ = runtime.StopContainer(stopCtx, container.ID)
82+ cancelStop()
83+
84+ removeCtx, cancelRemove := context.WithTimeout(context.Background(), 20*time.Second)
85+ _ = runtime.RemoveContainer(removeCtx, container.ID, true)
86+ cancelRemove()
74 })87 })
75- require.NoError(t, runtime.StartContainer(ctx, container.ID))88+ startCtx, cancelStart := context.WithTimeout(ctx, 2*time.Minute)
89+ defer cancelStart()
90+ require.NoError(t, runtime.StartContainer(startCtx, container.ID))
76 91
77 require.NoError(t, waitForPostgresReady(ctx, runtime, container.ID, 60*time.Second))92 require.NoError(t, waitForPostgresReady(ctx, runtime, container.ID, 60*time.Second))
78 require.NoError(t, seedPostgresData(ctx, runtime, container.ID))93 require.NoError(t, seedPostgresData(ctx, runtime, container.ID))
M internal/usecase/backup/service.go +4 -1
@@ -211,7 +211,10 @@ func selectDatabase(dbs []domain.DBInfo, requested string) (domain.DBInfo, error
211211 }
212212
213213 if requested == "" {
214- return dbs[0], nil
214+ if len(dbs) == 1 {
215+ return dbs[0], nil
216+ }
217+ return domain.DBInfo{}, fmt.Errorf("multiple database attachments detected; please specify one")
215218 }
216219
217220 for _, db := range dbs {
@@ -211,7 +211,10 @@ func selectDatabase(dbs []domain.DBInfo, requested string) (domain.DBInfo, error
211 }211 }
212 212
213 if requested == "" {213 if requested == "" {
214- return dbs[0], nil214+ if len(dbs) == 1 {
215+ return dbs[0], nil
216+ }
217+ return domain.DBInfo{}, fmt.Errorf("multiple database attachments detected; please specify one")
215 }218 }
216 219
217 for _, db := range dbs {220 for _, db := range dbs {
M internal/usecase/backup/service_test.go +19 -3
@@ -48,7 +48,7 @@ func TestService_RunBackup_Postgres(t *testing.T) {
4848 return false
4949 }
5050 return bytes.Contains([]byte(cmd[2]), []byte("pg_dump -Fc")) && bytes.Contains([]byte(cmd[2]), []byte(" > "))
51- })).Return(&outiface.ExecResult{ExitCode: 0}, nil)
51+ })).Return(&outiface.ExecResult{ExitCode: 0, Stdout: []byte("backup-data")}, nil)
5252
5353 runtime.EXPECT().CopyFromContainer(mock.Anything, "db123", mock.MatchedBy(func(path string) bool {
5454 return path != ""
@@ -65,8 +65,8 @@ func TestService_RunBackup_Postgres(t *testing.T) {
6565 domain.BackupSchedule(""),
6666 mock.Anything,
6767 mock.MatchedBy(func(r io.Reader) bool {
68- data, err := io.ReadAll(r)
69- return err == nil && string(data) == "backup-data"
68+ data, _ := io.ReadAll(r)
69+ return string(data) == "backup-data"
7070 }),
7171 ).Return("/tmp/backup.bak", nil)
7272
@@ -79,3 +79,19 @@ func TestService_RunBackup_Postgres(t *testing.T) {
7979 assert.Equal(t, "/tmp/backup.bak", result.Job.FilePath)
8080 assert.Equal(t, int64(len("backup-data")), result.Job.SizeBytes)
8181 }
82+
83+func TestSelectDatabaseRequiresExplicitNameWhenMultipleDetected(t *testing.T) {
84+ db, err := selectDatabase([]domain.DBInfo{
85+ {Name: "postgres"},
86+ {Name: "analytics"},
87+ }, "")
88+ require.Error(t, err)
89+ assert.Contains(t, err.Error(), "multiple database attachments detected")
90+ assert.Equal(t, domain.DBInfo{}, db)
91+}
92+
93+func TestSelectDatabaseAutoSelectsOnlyDatabaseWhenUnspecified(t *testing.T) {
94+ db, err := selectDatabase([]domain.DBInfo{{Name: "postgres"}}, "")
95+ require.NoError(t, err)
96+ assert.Equal(t, "postgres", db.Name)
97+}
@@ -48,7 +48,7 @@ func TestService_RunBackup_Postgres(t *testing.T) {
48 return false48 return false
49 }49 }
50 return bytes.Contains([]byte(cmd[2]), []byte("pg_dump -Fc")) && bytes.Contains([]byte(cmd[2]), []byte(" > "))50 return bytes.Contains([]byte(cmd[2]), []byte("pg_dump -Fc")) && bytes.Contains([]byte(cmd[2]), []byte(" > "))
51- })).Return(&outiface.ExecResult{ExitCode: 0}, nil)51+ })).Return(&outiface.ExecResult{ExitCode: 0, Stdout: []byte("backup-data")}, nil)
52 52
53 runtime.EXPECT().CopyFromContainer(mock.Anything, "db123", mock.MatchedBy(func(path string) bool {53 runtime.EXPECT().CopyFromContainer(mock.Anything, "db123", mock.MatchedBy(func(path string) bool {
54 return path != ""54 return path != ""
@@ -65,8 +65,8 @@ func TestService_RunBackup_Postgres(t *testing.T) {
65 domain.BackupSchedule(""),65 domain.BackupSchedule(""),
66 mock.Anything,66 mock.Anything,
67 mock.MatchedBy(func(r io.Reader) bool {67 mock.MatchedBy(func(r io.Reader) bool {
68- data, err := io.ReadAll(r)68+ data, _ := io.ReadAll(r)
69- return err == nil && string(data) == "backup-data"69+ return string(data) == "backup-data"
70 }),70 }),
71 ).Return("/tmp/backup.bak", nil)71 ).Return("/tmp/backup.bak", nil)
72 72
@@ -79,3 +79,19 @@ func TestService_RunBackup_Postgres(t *testing.T) {
79 assert.Equal(t, "/tmp/backup.bak", result.Job.FilePath)79 assert.Equal(t, "/tmp/backup.bak", result.Job.FilePath)
80 assert.Equal(t, int64(len("backup-data")), result.Job.SizeBytes)80 assert.Equal(t, int64(len("backup-data")), result.Job.SizeBytes)
81 }81 }
82+
83+func TestSelectDatabaseRequiresExplicitNameWhenMultipleDetected(t *testing.T) {
84+ db, err := selectDatabase([]domain.DBInfo{
85+ {Name: "postgres"},
86+ {Name: "analytics"},
87+ }, "")
88+ require.Error(t, err)
89+ assert.Contains(t, err.Error(), "multiple database attachments detected")
90+ assert.Equal(t, domain.DBInfo{}, db)
91+}
92+
93+func TestSelectDatabaseAutoSelectsOnlyDatabaseWhenUnspecified(t *testing.T) {
94+ db, err := selectDatabase([]domain.DBInfo{{Name: "postgres"}}, "")
95+ require.NoError(t, err)
96+ assert.Equal(t, "postgres", db.Name)
97+}
M internal/usecase/cron/scheduler.go +38 -3
@@ -2,6 +2,7 @@ package cron
22
33 import (
44 "context"
5+ "errors"
56 "fmt"
67 "sync"
78 "sync/atomic"
@@ -22,6 +23,20 @@ type Scheduler struct {
2223 started atomic.Bool
2324 }
2425
26+// Option configures a Scheduler.
27+type Option func(*Scheduler)
28+
29+var ErrJobRunning = errors.New("schedule is currently running")
30+
31+// WithMaxJobs sets the maximum number of jobs run concurrently.
32+func WithMaxJobs(n int) Option {
33+ return func(s *Scheduler) {
34+ if n > 0 {
35+ s.maxJobs = n
36+ }
37+ }
38+}
39+
2540 type entry struct {
2641 id string
2742 name string
@@ -33,8 +48,8 @@ type entry struct {
3348 }
3449
3550 // NewScheduler creates a scheduler instance.
36-func NewScheduler(log zerowrap.Logger) *Scheduler {
37- return &Scheduler{
51+func NewScheduler(log zerowrap.Logger, opts ...Option) *Scheduler {
52+ s := &Scheduler{
3853 entries: make(map[string]*entry),
3954 stopCh: make(chan struct{}),
4055 log: log,
@@ -43,6 +58,14 @@ func NewScheduler(log zerowrap.Logger) *Scheduler {
4358 return time.Now().UTC()
4459 },
4560 }
61+
62+ for _, opt := range opts {
63+ if opt != nil {
64+ opt(s)
65+ }
66+ }
67+
68+ return s
4669 }
4770
4871 // Add registers a new scheduled job.
@@ -78,10 +101,18 @@ func (s *Scheduler) Add(id, name string, sched domain.CronSchedule, job func(ctx
78101 }
79102
80103 // Remove unregisters a scheduled job.
81-func (s *Scheduler) Remove(id string) {
104+func (s *Scheduler) Remove(id string) error {
82105 s.mu.Lock()
83106 defer s.mu.Unlock()
107+ e := s.entries[id]
108+ if e == nil {
109+ return nil
110+ }
111+ if e.running.Load() {
112+ return fmt.Errorf("%w: %s", ErrJobRunning, id)
113+ }
84114 delete(s.entries, id)
115+ return nil
85116 }
86117
87118 // Start begins the scheduler loop.
@@ -155,17 +186,21 @@ func (s *Scheduler) runDue(ctx context.Context) {
155186 now := s.nowFn()
156187 entries := s.dueEntries(now)
157188 sem := make(chan struct{}, s.maxJobs)
189+ var wg sync.WaitGroup
158190 for _, e := range entries {
191+ wg.Add(1)
159192 sem <- struct{}{}
160193
161194 e := e
162195 go func() {
196+ defer wg.Done()
163197 defer func() { <-sem }()
164198 if err := s.executeEntry(ctx, e); err != nil {
165199 s.log.Warn().Err(err).Str("schedule_id", e.id).Msg("scheduled job failed")
166200 }
167201 }()
168202 }
203+ wg.Wait()
169204 }
170205
171206 func (s *Scheduler) executeEntry(ctx context.Context, e *entry) error {
@@ -2,6 +2,7 @@ package cron
2 2
3 import (3 import (
4 "context"4 "context"
5+ "errors"
5 "fmt"6 "fmt"
6 "sync"7 "sync"
7 "sync/atomic"8 "sync/atomic"
@@ -22,6 +23,20 @@ type Scheduler struct {
22 started atomic.Bool23 started atomic.Bool
23 }24 }
24 25
26+// Option configures a Scheduler.
27+type Option func(*Scheduler)
28+
29+var ErrJobRunning = errors.New("schedule is currently running")
30+
31+// WithMaxJobs sets the maximum number of jobs run concurrently.
32+func WithMaxJobs(n int) Option {
33+ return func(s *Scheduler) {
34+ if n > 0 {
35+ s.maxJobs = n
36+ }
37+ }
38+}
39+
25 type entry struct {40 type entry struct {
26 id string41 id string
27 name string42 name string
@@ -33,8 +48,8 @@ type entry struct {
33 }48 }
34 49
35 // NewScheduler creates a scheduler instance.50 // NewScheduler creates a scheduler instance.
36-func NewScheduler(log zerowrap.Logger) *Scheduler {51+func NewScheduler(log zerowrap.Logger, opts ...Option) *Scheduler {
37- return &Scheduler{52+ s := &Scheduler{
38 entries: make(map[string]*entry),53 entries: make(map[string]*entry),
39 stopCh: make(chan struct{}),54 stopCh: make(chan struct{}),
40 log: log,55 log: log,
@@ -43,6 +58,14 @@ func NewScheduler(log zerowrap.Logger) *Scheduler {
43 return time.Now().UTC()58 return time.Now().UTC()
44 },59 },
45 }60 }
61+
62+ for _, opt := range opts {
63+ if opt != nil {
64+ opt(s)
65+ }
66+ }
67+
68+ return s
46 }69 }
47 70
48 // Add registers a new scheduled job.71 // Add registers a new scheduled job.
@@ -78,10 +101,18 @@ func (s *Scheduler) Add(id, name string, sched domain.CronSchedule, job func(ctx
78 }101 }
79 102
80 // Remove unregisters a scheduled job.103 // Remove unregisters a scheduled job.
81-func (s *Scheduler) Remove(id string) {104+func (s *Scheduler) Remove(id string) error {
82 s.mu.Lock()105 s.mu.Lock()
83 defer s.mu.Unlock()106 defer s.mu.Unlock()
107+ e := s.entries[id]
108+ if e == nil {
109+ return nil
110+ }
111+ if e.running.Load() {
112+ return fmt.Errorf("%w: %s", ErrJobRunning, id)
113+ }
84 delete(s.entries, id)114 delete(s.entries, id)
115+ return nil
85 }116 }
86 117
87 // Start begins the scheduler loop.118 // Start begins the scheduler loop.
@@ -155,17 +186,21 @@ func (s *Scheduler) runDue(ctx context.Context) {
155 now := s.nowFn()186 now := s.nowFn()
156 entries := s.dueEntries(now)187 entries := s.dueEntries(now)
157 sem := make(chan struct{}, s.maxJobs)188 sem := make(chan struct{}, s.maxJobs)
189+ var wg sync.WaitGroup
158 for _, e := range entries {190 for _, e := range entries {
191+ wg.Add(1)
159 sem <- struct{}{}192 sem <- struct{}{}
160 193
161 e := e194 e := e
162 go func() {195 go func() {
196+ defer wg.Done()
163 defer func() { <-sem }()197 defer func() { <-sem }()
164 if err := s.executeEntry(ctx, e); err != nil {198 if err := s.executeEntry(ctx, e); err != nil {
165 s.log.Warn().Err(err).Str("schedule_id", e.id).Msg("scheduled job failed")199 s.log.Warn().Err(err).Str("schedule_id", e.id).Msg("scheduled job failed")
166 }200 }
167 }()201 }()
168 }202 }
203+ wg.Wait()
169 }204 }
170 205
171 func (s *Scheduler) executeEntry(ctx context.Context, e *entry) error {206 func (s *Scheduler) executeEntry(ctx context.Context, e *entry) error {
M internal/usecase/cron/scheduler_test.go +32 -0
@@ -75,6 +75,38 @@ func TestSchedulerAddListAndRunNow(t *testing.T) {
7575 entries = s.List()
7676 require.Len(t, entries, 1)
7777 assert.Equal(t, now, entries[0].LastRun)
78+ assert.Equal(t, time.Date(2026, 2, 8, 2, 0, 0, 0, time.UTC), entries[0].NextRun)
79+}
80+
81+func TestSchedulerWithMaxJobsOption(t *testing.T) {
82+ s := NewScheduler(zerowrap.Default(), WithMaxJobs(9))
83+ assert.Equal(t, 9, s.maxJobs)
84+}
85+
86+func TestSchedulerRemoveRejectsRunningJob(t *testing.T) {
87+ s := NewScheduler(zerowrap.Default())
88+
89+ started := make(chan struct{})
90+ release := make(chan struct{})
91+ err := s.Add("backup-hourly", "hourly backup", domain.CronSchedule{Preset: domain.ScheduleHourly}, func(context.Context) error {
92+ close(started)
93+ <-release
94+ return nil
95+ })
96+ require.NoError(t, err)
97+
98+ runErrCh := make(chan error, 1)
99+ go func() {
100+ runErrCh <- s.RunNow(context.Background(), "backup-hourly")
101+ }()
102+
103+ <-started
104+ err = s.Remove("backup-hourly")
105+ require.Error(t, err)
106+ assert.ErrorIs(t, err, ErrJobRunning)
107+
108+ close(release)
109+ require.NoError(t, <-runErrCh)
78110 }
79111
80112 func TestSchedulerRunNowRejectsConcurrentRun(t *testing.T) {
@@ -75,6 +75,38 @@ func TestSchedulerAddListAndRunNow(t *testing.T) {
75 entries = s.List()75 entries = s.List()
76 require.Len(t, entries, 1)76 require.Len(t, entries, 1)
77 assert.Equal(t, now, entries[0].LastRun)77 assert.Equal(t, now, entries[0].LastRun)
78+ assert.Equal(t, time.Date(2026, 2, 8, 2, 0, 0, 0, time.UTC), entries[0].NextRun)
79+}
80+
81+func TestSchedulerWithMaxJobsOption(t *testing.T) {
82+ s := NewScheduler(zerowrap.Default(), WithMaxJobs(9))
83+ assert.Equal(t, 9, s.maxJobs)
84+}
85+
86+func TestSchedulerRemoveRejectsRunningJob(t *testing.T) {
87+ s := NewScheduler(zerowrap.Default())
88+
89+ started := make(chan struct{})
90+ release := make(chan struct{})
91+ err := s.Add("backup-hourly", "hourly backup", domain.CronSchedule{Preset: domain.ScheduleHourly}, func(context.Context) error {
92+ close(started)
93+ <-release
94+ return nil
95+ })
96+ require.NoError(t, err)
97+
98+ runErrCh := make(chan error, 1)
99+ go func() {
100+ runErrCh <- s.RunNow(context.Background(), "backup-hourly")
101+ }()
102+
103+ <-started
104+ err = s.Remove("backup-hourly")
105+ require.Error(t, err)
106+ assert.ErrorIs(t, err, ErrJobRunning)
107+
108+ close(release)
109+ require.NoError(t, <-runErrCh)
78 }110 }
79 111
80 func TestSchedulerRunNowRejectsConcurrentRun(t *testing.T) {112 func TestSchedulerRunNowRejectsConcurrentRun(t *testing.T) {

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 7

Tags 106

↑↓ moveenter openesc close