[locker] add Close method

Author: capcom6

internal/worker/locker/locker.go

@@ -5,9 +5,17 @@ import (
 	"errors"
 )
 
+// ErrLockNotAcquired is returned when a lock cannot be acquired within the configured timeout.
 var ErrLockNotAcquired = errors.New("lock not acquired")
 
 type Locker interface {
+	// AcquireLock attempts to acquire a lock for the given key.
+	// Returns ErrLockNotAcquired if the lock cannot be acquired within the configured timeout.
 	AcquireLock(ctx context.Context, key string) error
+	// ReleaseLock releases a previously acquired lock for the given key.
+	// Returns an error if the lock was not held by this instance.
 	ReleaseLock(ctx context.Context, key string) error
+
+	// Close releases any held locks.
+	Close() error
 }

internal/worker/locker/module.go

@@ -1,6 +1,7 @@
 package locker
 
 import (
+	"context"
 	"database/sql"
 
 	"github.com/go-core-fx/logger"
@@ -16,5 +17,15 @@ func Module() fx.Option {
 		fx.Provide(func(db *sql.DB) Locker {
 			return NewMySQLLocker(db, "worker:", timeoutSeconds)
 		}),
+		fx.Invoke(func(locker Locker, lc fx.Lifecycle) {
+			lc.Append(fx.Hook{
+				OnStart: func(_ context.Context) error {
+					return nil
+				},
+				OnStop: func(_ context.Context) error {
+					return locker.Close()
+				},
+			})
+		}),
 	)
 }

internal/worker/locker/mysql.go

@@ -5,19 +5,21 @@ import (
 	"database/sql"
 	"fmt"
 	"sync"
+	"time"
 )
 
 type mySQLLocker struct {
 	db *sql.DB
 
 	prefix  string
-	timeout uint
+	timeout time.Duration
 
 	mu    sync.Mutex
 	conns map[string]*sql.Conn
 }
 
-func NewMySQLLocker(db *sql.DB, prefix string, timeout uint) Locker {
+// NewMySQLLocker creates a new MySQL-based distributed locker.
+func NewMySQLLocker(db *sql.DB, prefix string, timeout time.Duration) Locker {
 	return &mySQLLocker{
 		db: db,
 
@@ -39,7 +41,7 @@ func (m *mySQLLocker) AcquireLock(ctx context.Context, key string) error {
 	}
 
 	var res sql.NullInt64
-	if err := conn.QueryRowContext(ctx, "SELECT GET_LOCK(?, ?)", name, m.timeout).Scan(&res); err != nil {
+	if err := conn.QueryRowContext(ctx, "SELECT GET_LOCK(?, ?)", name, m.timeout.Seconds()).Scan(&res); err != nil {
 		_ = conn.Close()
 		return fmt.Errorf("failed to get lock: %w", err)
 	}
@@ -85,4 +87,18 @@ func (m *mySQLLocker) ReleaseLock(ctx context.Context, key string) error {
 	return nil
 }
 
+// Close closes all remaining pinned connections.
+// Should be called during shutdown to clean up resources.
+func (m *mySQLLocker) Close() error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for key, conn := range m.conns {
+		if conn != nil {
+			_ = conn.Close()
+		}
+		delete(m.conns, key)
+	}
+	return nil
+}
+
 var _ Locker = (*mySQLLocker)(nil)