perf(1password): warm session once, then resolve apps concurrently

cli/execute.go

@@ -71,6 +71,7 @@ func newRunCmd(fc fileConfig) *cobra.Command {
 	shared := &sharedFlags{}
 	var secretResolver string
 	var composeFiles []string
+	var concurrency int
 
 	cmd := &cobra.Command{
 		Use:   "run [app-path...] --secret-resolver <name> [flags] -- <command>...",
@@ -91,6 +92,7 @@ applies on top.`,
 	addSharedFlags(cmd, shared)
 	cmd.Flags().StringVar(&secretResolver, "secret-resolver", "", "secret backend: "+resolve.SecretResolverNames())
 	cmd.Flags().StringArrayVar(&composeFiles, "compose-file", nil, "docker compose file COMPOSE_FILE points at, relative to --root; repeatable, later files win (default \"docker-compose.yml\")")
+	cmd.Flags().IntVar(&concurrency, "concurrency", run.DefaultConcurrency, "how many apps resolve their secrets at once")
 	resolverFor := bindSelectedSecretResolver(cmd, fc)
 
 	cmd.RunE = func(cmd *cobra.Command, args []string) error {
@@ -122,6 +124,7 @@ applies on top.`,
 			Composer:     compose.NewComposer(),
 			Project:      shared.project(),
 			ComposeFiles: composeFiles,
+			Concurrency:  concurrency,
 		})
 		return runner.Run(cmd.Context(), run.Params{
 			Apps:        apps,

cli/resolvers/onepassword/onepassword.go

@@ -14,11 +14,38 @@ import (
 	"github.com/spf13/pflag"
 )
 
-// opMu serializes `op` invocations. Callers resolve apps concurrently, but the op
-// CLI unlocks via a biometric prompt per invocation until its desktop-app session
-// is cached; firing many at once triggers a prompt storm. Running them one at a
-// time lets the first prompt cache the session so the rest reuse it silently.
-var opMu sync.Mutex
+// opWarm gates the first `op` invocation so it runs alone. Callers resolve apps
+// concurrently, but the op CLI unlocks via a biometric prompt per invocation
+// until its desktop-app session is cached; firing many at once against a cold
+// session triggers a prompt storm. Serializing until the first call succeeds lets
+// that one prompt cache the session, after which the rest run concurrently and
+// reuse it silently. Only the very first call pays the serial cost.
+var opWarm warmGate
+
+// warmGate serializes callers until the first op invocation succeeds, then lets
+// every later caller run concurrently.
+type warmGate struct {
+	mu     sync.Mutex
+	warmed bool
+}
+
+// run executes exec under the warm gate: once the session is warm it runs
+// straight through, otherwise it holds the lock across the call so a cold session
+// only ever prompts once, marking the gate warm on the first success so queued
+// callers drain into concurrent mode.
+func (g *warmGate) run(exec func() error) error {
+	g.mu.Lock()
+	if g.warmed {
+		g.mu.Unlock()
+		return exec()
+	}
+	defer g.mu.Unlock()
+	err := exec()
+	if err == nil {
+		g.warmed = true
+	}
+	return err
+}
 
 func init() {
 	cli.RegisterSecretResolver(cli.SecretResolverCommand{
@@ -66,9 +93,7 @@ func (o onePassword) Resolve(ctx context.Context, names []string) (map[string]st
 	var stdout, stderr bytes.Buffer
 	cmd.Stdout = &stdout
 	cmd.Stderr = &stderr
-	opMu.Lock()
-	err := cmd.Run()
-	opMu.Unlock()
+	err := opWarm.run(cmd.Run)
 
 	return o.pick(opResult{err: err, stdout: stdout.Bytes(), stderr: stderr.String()}, names)
 }

cli/resolvers/onepassword/onepassword_test.go

@@ -3,7 +3,10 @@ package onepassword
 import (
 	"encoding/json"
 	"errors"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/harrisoncramer/ultra/cli"
 
@@ -56,3 +59,47 @@ func TestItemNotFound(t *testing.T) {
 	assert.False(t, itemNotFound(`"Employee" isn't a vault in this account.`))
 	assert.False(t, itemNotFound("connecting to desktop app: timed out"))
 }
+
+// Until the first call succeeds the gate serializes, so a cold session prompts
+// only once; a warmed gate then lets calls overlap. peakConcurrency reaching >1
+// only after the warm-up proves both halves.
+func TestWarmGateSerializesUntilWarm(t *testing.T) {
+	var g warmGate
+
+	// Warm the gate with one successful call. A concurrent burst launched before
+	// this would have been serialized behind it.
+	require.NoError(t, g.run(func() error { return nil }))
+
+	const workers = 8
+	var inFlight int32
+	block := make(chan struct{})
+	var wg sync.WaitGroup
+	for range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_ = g.run(func() error {
+				atomic.AddInt32(&inFlight, 1)
+				<-block
+				return nil
+			})
+		}()
+	}
+
+	// Wait until every worker is parked inside exec at once; a gate that
+	// serialized would never let inFlight reach workers, so this would hang and
+	// fail the test via timeout rather than pass falsely.
+	require.Eventually(t, func() bool {
+		return atomic.LoadInt32(&inFlight) == workers
+	}, time.Second, time.Millisecond)
+	close(block)
+	wg.Wait()
+}
+
+// A cold gate whose first call fails stays cold, so the next caller is still
+// serialized rather than joining a false warm.
+func TestWarmGateStaysColdOnFailure(t *testing.T) {
+	var g warmGate
+	require.Error(t, g.run(func() error { return errors.New("op unreachable") }))
+	assert.False(t, g.warmed)
+}

docs/reference/ultra_run.md

@@ -24,6 +24,7 @@ ultra run [app-path...] --secret-resolver <name> [flags] -- <command>...
 
 ```
       --compose-file stringArray   docker compose file COMPOSE_FILE points at, relative to --root; repeatable, later files win (default "docker-compose.yml")
+      --concurrency int            how many apps resolve their secrets at once (default 16)
       --config-dir string          config package directory under each app path (e.g. pkg/config) (default "config")
       --config-file string         path to the ultra config file (default .ultra.toml under --root)
   -h, --help                       help for run

internal/run/run.go

@@ -21,10 +21,12 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
-// maxConcurrentApps bounds how many apps resolve at once, so a project with many
-// apps doesn't spawn an unbounded number of resolver subprocesses (op, docker)
-// simultaneously.
-const maxConcurrentApps = 8
+// DefaultConcurrency bounds how many apps resolve at once when the caller does
+// not set one, so a project with many apps doesn't spawn an unbounded number of
+// resolver subprocesses (op, docker) simultaneously. It comfortably covers a
+// typical project's app count so every app resolves in one wave; --concurrency
+// tunes it per invocation.
+const DefaultConcurrency = 16
 
 // generator writes the combined compose override and reports the secret names
 // each app declares, independent of the secret store.
@@ -44,6 +46,7 @@ type Runner struct {
 	composer     composer
 	project      project.Project
 	composeFiles []string
+	concurrency  int
 }
 
 // NewRunnerParams are the dependencies and layout NewRunner needs.
@@ -55,19 +58,27 @@ type NewRunnerParams struct {
 	// a later file (e.g. a local override) wins over an earlier one; the generated
 	// secrets override is always layered last. Empty means "docker-compose.yml".
 	ComposeFiles []string
+	// Concurrency bounds how many apps resolve their secrets at once. Zero or
+	// negative means DefaultConcurrency.
+	Concurrency int
 }
 
-// NewRunner builds a Runner, defaulting the compose file.
+// NewRunner builds a Runner, defaulting the compose file and concurrency.
 func NewRunner(params NewRunnerParams) *Runner {
 	composeFiles := params.ComposeFiles
 	if len(composeFiles) == 0 {
 		composeFiles = []string{"docker-compose.yml"}
 	}
+	concurrency := params.Concurrency
+	if concurrency <= 0 {
+		concurrency = DefaultConcurrency
+	}
 	return &Runner{
 		generator:    params.Generator,
 		composer:     params.Composer,
 		project:      params.Project,
 		composeFiles: composeFiles,
+		concurrency:  concurrency,
 	}
 }
 
@@ -121,7 +132,7 @@ func (r *Runner) prepare(ctx context.Context, params Params) (*prepared, error)
 	results := make([]appResult, len(overrides))
 
 	g, ctx := errgroup.WithContext(ctx)
-	g.SetLimit(maxConcurrentApps)
+	g.SetLimit(r.concurrency)
 	for i, o := range overrides {
 		if len(o.Names) == 0 {
 			continue

internal/run/run_test.go

@@ -2,10 +2,13 @@ package run
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/harrisoncramer/ultra/internal/compose"
 	"github.com/harrisoncramer/ultra/internal/configreader"
@@ -232,3 +235,73 @@ func TestRunRejectsEmptyCommand(t *testing.T) {
 	require.Error(t, err, "an empty command must error, not panic on Command[0]")
 	assert.Contains(t, err.Error(), "no command to run")
 }
+
+// NewRunner falls back to DefaultConcurrency when the caller leaves it unset or
+// passes a non-positive value, and keeps a positive one.
+func TestNewRunnerConcurrencyDefault(t *testing.T) {
+	proj := project.Project{Root: t.TempDir(), ConfigDir: "config"}
+	base := NewRunnerParams{Composer: fakeComposer{}, Project: proj}
+
+	assert.Equal(t, DefaultConcurrency, NewRunner(base).concurrency)
+
+	base.Concurrency = -1
+	assert.Equal(t, DefaultConcurrency, NewRunner(base).concurrency)
+
+	base.Concurrency = 4
+	assert.Equal(t, 4, NewRunner(base).concurrency)
+}
+
+// concurrencyProbe records the peak number of resolvers running at once.
+type concurrencyProbe struct {
+	mu       *sync.Mutex
+	inFlight *int
+	peak     *int
+}
+
+func (p concurrencyProbe) Resolve(context.Context, []string) (map[string]string, error) {
+	p.mu.Lock()
+	*p.inFlight++
+	if *p.inFlight > *p.peak {
+		*p.peak = *p.inFlight
+	}
+	p.mu.Unlock()
+	time.Sleep(20 * time.Millisecond)
+	p.mu.Lock()
+	*p.inFlight--
+	p.mu.Unlock()
+	return nil, nil
+}
+
+// prepare never runs more resolvers at once than the configured concurrency,
+// while still running several concurrently.
+func TestPrepareRespectsConcurrency(t *testing.T) {
+	root := t.TempDir()
+	proj := project.Project{Root: root, ConfigDir: "config"}
+	const limit = 3
+	runner := NewRunner(NewRunnerParams{
+		Generator: gen.NewGenerator(gen.NewGeneratorParams{
+			Reader:   configreader.NewConfigReader(configreader.NewConfigReaderParams{Scanner: fakeScanner{names: []string{"SECRET"}}, Project: proj}),
+			Composer: fakeComposer{},
+			Project:  proj,
+		}),
+		Composer:    fakeComposer{},
+		Project:     proj,
+		Concurrency: limit,
+	})
+
+	var mu sync.Mutex
+	var inFlight, peak int
+	apps := make([]string, 12)
+	for i := range apps {
+		apps[i] = fmt.Sprintf("app%d", i)
+	}
+	_, err := runner.prepare(context.Background(), Params{
+		Apps: apps,
+		ResolverFor: func(string) resolve.SecretResolver {
+			return concurrencyProbe{mu: &mu, inFlight: &inFlight, peak: &peak}
+		},
+	})
+	require.NoError(t, err)
+	assert.LessOrEqual(t, peak, limit, "ran more resolvers at once than the limit")
+	assert.Greater(t, peak, 1, "resolvers should run concurrently")
+}