[mirror] a bluesky bot to post golang projects
Merge pull request #30 from till/fix-everything-omg
fix: stop leaking cleanup goroutines
+66
-32
+53
-29
internal/cmd/run.go
+53
-29
internal/cmd/run.go
···
44
44
45
45
// RunWithReconnect attempts to run the bot with automatic reconnection on failure
46
46
func RunWithReconnect(ctx context.Context, mc *minio.Client, cfg Config) error {
47
+
cacheClient := content.NewCacheClientS3(ctx, mc, cfg.CacheBucket)
48
+
49
+
cleanup := content.NewS3Cleanup(mc, cfg.CacheBucket)
50
+
cleanup.Start(ctx)
51
+
defer cleanup.Stop()
52
+
53
+
if err := content.Start(cfg.GitHubToken, cacheClient); err != nil {
54
+
return fmt.Errorf("failed to start service: %w", err)
55
+
}
56
+
47
57
for {
58
+
if err := ctx.Err(); err != nil {
59
+
return err
60
+
}
61
+
48
62
client, err := connectBluesky(ctx, cfg.Handle, cfg.AppKey)
49
63
if err != nil {
64
+
if ctx.Err() != nil {
65
+
return ctx.Err()
66
+
}
50
67
slog.Error("failed to connect to Bluesky", "error", err)
51
68
slog.Info("retrying connection", "delay", reconnectDelay)
52
-
time.Sleep(reconnectDelay)
69
+
if err := sleepCtx(ctx, reconnectDelay); err != nil {
70
+
return err
71
+
}
53
72
continue
54
73
}
55
74
56
-
c := bluesky.Client{
57
-
Client: client,
58
-
}
59
-
60
-
cacheClient := content.NewCacheClientS3(ctx, mc, cfg.CacheBucket)
61
-
62
-
// Initialize and start the cleanup handler
63
-
cleanup := content.NewS3Cleanup(mc, cfg.CacheBucket)
64
-
cleanup.Start(ctx)
65
-
defer cleanup.Stop()
75
+
c := bluesky.Client{Client: client}
76
+
runSession(ctx, c)
77
+
client.Close()
66
78
67
-
if err := content.Start(cfg.GitHubToken, cacheClient); err != nil {
68
-
slog.Error("failed to start service", "error", err)
69
-
client.Close()
70
-
time.Sleep(reconnectDelay)
71
-
continue
79
+
if err := sleepCtx(ctx, reconnectDelay); err != nil {
80
+
return err
72
81
}
82
+
}
83
+
}
73
84
74
-
// Run the main loop
75
-
for {
76
-
slog.DebugContext(ctx, "checking...")
77
-
if err := content.Do(ctx, c); err != nil {
78
-
if !errors.Is(err, content.ErrCouldNotContent) {
79
-
slog.Error("error during content check", "error", err)
80
-
client.Close()
81
-
time.Sleep(reconnectDelay)
82
-
break
83
-
}
84
-
slog.DebugContext(ctx, "backing off...")
85
+
// runSession runs the inner check loop until ctx is cancelled or content.Do
86
+
// returns a non-recoverable error. The caller is responsible for closing the
87
+
// bluesky client and reconnecting.
88
+
func runSession(ctx context.Context, c bluesky.Client) {
89
+
for {
90
+
slog.DebugContext(ctx, "checking...")
91
+
if err := content.Do(ctx, c); err != nil {
92
+
if !errors.Is(err, content.ErrCouldNotContent) {
93
+
slog.Error("error during content check", "error", err)
94
+
return
85
95
}
86
-
87
-
time.Sleep(checkInterval)
96
+
slog.DebugContext(ctx, "backing off...")
88
97
}
98
+
if err := sleepCtx(ctx, checkInterval); err != nil {
99
+
return
100
+
}
101
+
}
102
+
}
103
+
104
+
// sleepCtx sleeps for d, returning ctx.Err() if ctx is cancelled first.
105
+
func sleepCtx(ctx context.Context, d time.Duration) error {
106
+
t := time.NewTimer(d)
107
+
defer t.Stop()
108
+
select {
109
+
case <-ctx.Done():
110
+
return ctx.Err()
111
+
case <-t.C:
112
+
return nil
89
113
}
90
114
}
+13
-3
internal/content/s3_cleanup.go
+13
-3
internal/content/s3_cleanup.go
···
3
3
import (
4
4
"context"
5
5
"fmt"
6
+
"sync"
6
7
"time"
7
8
8
9
"github.com/minio/minio-go/v7"
···
17
18
cleanupInterval time.Duration
18
19
// stopCleanup is used to signal the cleanup routine to stop
19
20
stopCleanup chan struct{}
21
+
stopOnce sync.Once
20
22
}
21
23
22
24
// NewS3Cleanup creates a new S3 cleanup handler
···
34
36
go c.cleanupRoutine(ctx)
35
37
}
36
38
37
-
// Stop stops the background cleanup routine
39
+
// Stop stops the background cleanup routine. Safe to call multiple times.
38
40
func (c *S3Cleanup) Stop() {
39
-
close(c.stopCleanup)
41
+
c.stopOnce.Do(func() { close(c.stopCleanup) })
40
42
}
41
43
42
-
// cleanupRoutine periodically checks for and deletes expired objects
44
+
// cleanupRoutine periodically checks for and deletes expired objects.
45
+
// It runs once at startup before entering the ticker loop so restarts
46
+
// shorter than cleanupInterval do not skip cleanup entirely.
43
47
func (c *S3Cleanup) cleanupRoutine(ctx context.Context) {
44
48
ticker := time.NewTicker(c.cleanupInterval)
45
49
defer ticker.Stop()
46
50
51
+
if err := c.cleanupExpired(ctx); err != nil {
52
+
utils.LogErrorWithContext(ctx, fmt.Errorf("failed to cleanup expired objects: %w", err))
53
+
}
54
+
47
55
for {
48
56
select {
49
57
case <-ticker.C:
50
58
if err := c.cleanupExpired(ctx); err != nil {
51
59
utils.LogErrorWithContext(ctx, fmt.Errorf("failed to cleanup expired objects: %w", err))
52
60
}
61
+
case <-ctx.Done():
62
+
return
53
63
case <-c.stopCleanup:
54
64
return
55
65
}