initial

1 files changed, 110 insertions, 0 deletions

diff --git a/internal/cli/worker_pool.go b/internal/cli/worker_pool.go
new file mode 100644
index 0000000..fe44644
--- /dev/null
+++ b/internal/cli/worker_pool.go
@@ -0,0 +1,110 @@
+package cli
+
+import (
+	"context"
+	"errors"
+	"runtime"
+	"sync"
+	"sync/atomic"
+)
+
+// pool is a bounded worker pool for concurrent, independent tasks (§14). A
+// single caller feeds work via submit; up to concurrency tasks run in flight at
+// once, bounded by the sem token held from submit until the task returns.
+//
+// The pool is deliberately upload-agnostic: a task is any func(ctx) error, so
+// the same machinery serves push uploads today and prune deletes later. Task
+// wording, error identity, and any success side-effect live in the closure.
+//
+// submit is the only place that logs a per-action line, and it is only ever
+// called from one goroutine (the caller feeding work), so output needs no
+// locking; only the shared counters are guarded. Tasks run concurrently, so any
+// state a task closure touches beyond its own arguments is the caller's
+// responsibility to synchronize.
+type workerPool struct {
+	ctx    context.Context
+	cancel context.CancelCauseFunc
+	sem    chan struct{}
+	wg     sync.WaitGroup
+
+	// failFast cancels the pool on the first failure so in-flight tasks are
+	// abandoned and queued ones never start (the push-asset policy, §8.3). When
+	// false, every submitted task runs and all failures are collected (the
+	// commit/prune-delete policy).
+	failFast bool
+
+	mtx       sync.Mutex
+	errs      []error
+	succeeded int64
+}
+
+// newWorkerPool creates a new worker pool to process tasks concurrently. If `concurrency`
+// is zero, the number of parallel jobs is aut-detected based on the number of CPUs.
+func newWorkerPool(ctx context.Context, concurrency uint, failFast bool) *workerPool {
+	if concurrency == 0 {
+		concurrency = 2 * uint(runtime.NumCPU())
+	}
+	ctx, cancel := context.WithCancelCause(ctx)
+	return &workerPool{
+		ctx:      ctx,
+		cancel:   cancel,
+		sem:      make(chan struct{}, concurrency),
+		failFast: failFast,
+	}
+}
+
+// submit blocks until a worker slot is free, then runs task in the background.
+// It reports false when the pool has been cancelled (fail-fast or the parent
+// context), signalling the caller to stop feeding work. action is the text that
+// will be logged before the task is executed.
+func (p *workerPool) submit(action string, task func(ctx context.Context) error) bool {
+	select {
+	case p.sem <- struct{}{}:
+	case <-p.ctx.Done():
+		return false
+	}
+
+	logAction(action)
+
+	p.wg.Add(1)
+	go func() {
+		defer p.wg.Done()
+		defer func() { <-p.sem }()
+
+		if err := task(p.ctx); err != nil {
+			if !errors.Is(err, context.Canceled) {
+				p.mtx.Lock()
+				p.errs = append(p.errs, err)
+				p.mtx.Unlock()
+			}
+			if p.failFast {
+				p.cancel(CancellationError("operation aborted"))
+			}
+			return
+		}
+		atomic.AddInt64(&p.succeeded, 1)
+	}()
+	return true
+}
+
+// cancelled reports whether the pool's context is already done (fail-fast
+// triggered or the parent context cancelled). A feeder loop can poll this to
+// stop producing/buffering work promptly, instead of only discovering the
+// cancellation on its next submit.
+func (p *workerPool) cancelled() bool {
+	select {
+	case <-p.ctx.Done():
+		return true
+	default:
+		return false
+	}
+}
+
+// wait blocks until all in-flight tasks finish, then releases the pool's
+// context. It returns the number of succeeded tasks and the errors the
+// workers collected.
+func (p *workerPool) wait() (int, []error) {
+	p.wg.Wait()
+	p.cancel(nil)
+	return int(p.succeeded), p.errs
+}