gadfly/cmd/gadfly/lenssem.go

package main

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"syscall"
	"time"
)

// lensSem is the PROVIDER-WIDE lens-permit pool. Historically each model's binary
// throttled its own lenses in-process (GADFLY_LENS_CONCURRENCY) while entrypoint.sh
// separately capped how many MODELS ran at once — two multiplicative gates in
// different processes. That let a model hold its whole slot until its last lens
// finished, stalling the next model even with idle lens capacity.
//
// Instead, entrypoint.sh now runs all of a provider's models concurrently and
// seeds ONE directory of N permit files per provider (N = the provider's lens
// budget). Every model process in that lane draws from the same pool: a lens pass
// (review + recheck) acquires a permit before it runs and releases it after, so a
// model winding down immediately yields its freed permits to another model's
// queued lenses. Permits are held with flock, which the kernel drops when the
// holding process exits — so a killed/crashed model frees its permits for free.
type lensSem struct {
	dir  string
	size int
}

// lensSemPollInterval is how often a blocked acquirer re-sweeps the permit files.
// Lens passes run for many seconds to minutes, so a coarse poll adds negligible
// latency while keeping the mechanism a few lines of stdlib (no IPC primitives).
const lensSemPollInterval = 150 * time.Millisecond

// activeLensSem returns the shared semaphore configured by entrypoint.sh, or nil
// when it isn't set (standalone/local runs, tests, or an older entrypoint) — in
// which case runSpecialists falls back to the in-process fanout limit alone, i.e.
// the pre-existing per-model behavior.
func activeLensSem() *lensSem {
	dir := os.Getenv("GADFLY_LENS_SEM_DIR")
	if dir == "" {
		return nil
	}
	n := envInt("GADFLY_LENS_SEM_SIZE", 0)
	if n < 1 {
		// A dir was requested but the size is missing/blank/invalid. Rather than
		// silently run unthrottled, say so on stderr — it's almost certainly a
		// misconfiguration in entrypoint.sh (the two are set together).
		fmt.Fprintf(os.Stderr, "gadfly: GADFLY_LENS_SEM_DIR set but GADFLY_LENS_SEM_SIZE=%q invalid; lenses run unthrottled\n", os.Getenv("GADFLY_LENS_SEM_SIZE"))
		return nil
	}
	return &lensSem{dir: dir, size: n}
}

// acquire blocks until a permit is free (or ctx is done) and returns a release
// func. It FAILS OPEN: if the pool directory is structurally unusable (missing,
// unwritable, or on a filesystem without flock) it logs once and returns a no-op
// release with a nil error, so the lens runs UNTHROTTLED rather than hanging the
// whole review forever — this is an advisory reviewer, a slightly oversubscribed
// backend beats a stuck one. Only a full-but-healthy pool actually blocks; on ctx
// cancellation it returns a no-op release plus ctx.Err() so the caller can surface
// the lens as "did not run" instead of leaking a permit.
func (s *lensSem) acquire(ctx context.Context) (func(), error) {
	for {
		release, ok, err := s.tryAcquire()
		if ok {
			return release, nil
		}
		if err != nil {
			fmt.Fprintf(os.Stderr, "gadfly: lens permit pool %q unusable (%v); proceeding without a permit\n", s.dir, err)
			return func() {}, nil
		}
		timer := time.NewTimer(lensSemPollInterval)
		select {
		case <-ctx.Done():
			timer.Stop() // don't leak the timer when we bail on cancellation
			return func() {}, ctx.Err()
		case <-timer.C:
		}
	}
}

// tryAcquire makes one non-blocking sweep over the permit files. It returns a
// release func for the first one it locks; otherwise it distinguishes a healthy
// FULL pool (some permit was openable but flock-busy → ok=false, err=nil → keep
// polling) from a structurally BROKEN pool (no permit was even acquirable and none
// was merely busy → err set → caller fails open). Permit files are created lazily
// (entrypoint.sh only guarantees the directory exists). Closing the *os.File in
// the returned func drops the flock (the lock lives on the open file description).
func (s *lensSem) tryAcquire() (func(), bool, error) {
	sawBusy := false
	var structural error
	for i := 0; i < s.size; i++ {
		path := filepath.Join(s.dir, fmt.Sprintf("permit.%d", i))
		f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0o600)
		if err != nil {
			structural = err // e.g. the pool dir is gone or unwritable
			continue
		}
		if err := syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB); err == nil {
			return func() { f.Close() }, true, nil
		} else if errors.Is(err, syscall.EWOULDBLOCK) {
			sawBusy = true // this permit is held by someone else — pool has capacity
			f.Close()
		} else {
			structural = err // flock unsupported / other → not a transient "busy"
			f.Close()
		}
	}
	if sawBusy {
		return nil, false, nil // full but healthy: another lens will free a permit
	}
	return nil, false, structural
}