From 6746c8f4de9540dc322a756d57a94b20af1cf775 Mon Sep 17 00:00:00 2001 From: Harrison Cramer Date: Tue, 21 Jul 2026 16:23:19 -0400 Subject: [PATCH 1/2] perf(1password): warm session once, then resolve apps concurrently --- cli/resolvers/onepassword/onepassword.go | 41 ++++++++++++---- cli/resolvers/onepassword/onepassword_test.go | 47 +++++++++++++++++++ 2 files changed, 80 insertions(+), 8 deletions(-) diff --git a/cli/resolvers/onepassword/onepassword.go b/cli/resolvers/onepassword/onepassword.go index 19420f1..476911d 100644 --- a/cli/resolvers/onepassword/onepassword.go +++ b/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) } diff --git a/cli/resolvers/onepassword/onepassword_test.go b/cli/resolvers/onepassword/onepassword_test.go index aae449e..33b8aed 100644 --- a/cli/resolvers/onepassword/onepassword_test.go +++ b/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) +} From 9b84bd76236a709607071a8e05b2cb610ea99686 Mon Sep 17 00:00:00 2001 From: Harrison Cramer Date: Tue, 21 Jul 2026 16:39:08 -0400 Subject: [PATCH 2/2] feat(run): add --concurrency flag, raise default resolve concurrency to 16 --- cli/execute.go | 3 ++ docs/reference/ultra_run.md | 1 + internal/run/run.go | 23 +++++++++--- internal/run/run_test.go | 73 +++++++++++++++++++++++++++++++++++++ 4 files changed, 94 insertions(+), 6 deletions(-) diff --git a/cli/execute.go b/cli/execute.go index e920f4d..5f8808f 100644 --- a/cli/execute.go +++ b/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 [flags] -- ...", @@ -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, diff --git a/docs/reference/ultra_run.md b/docs/reference/ultra_run.md index d1cab7b..00f7a75 100644 --- a/docs/reference/ultra_run.md +++ b/docs/reference/ultra_run.md @@ -24,6 +24,7 @@ ultra run [app-path...] --secret-resolver [flags] -- ... ``` --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 diff --git a/internal/run/run.go b/internal/run/run.go index d54b8eb..af04ecc 100644 --- a/internal/run/run.go +++ b/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 diff --git a/internal/run/run_test.go b/internal/run/run_test.go index aba8165..b6bad5b 100644 --- a/internal/run/run_test.go +++ b/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") +}