From 6c098873a41f19fedd75eaa7b32d87f00127f4de Mon Sep 17 00:00:00 2001 From: Julius Park Date: Mon, 27 Jul 2026 15:28:07 -0400 Subject: [PATCH 1/5] initial commit --- .github/workflows/build.yml | 12 ++++- cmd/hatchet-loadtest/timing.go | 98 +++++++++++++++++++++++++++------- 2 files changed, 90 insertions(+), 20 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 081342acc8..3001129206 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -257,11 +257,21 @@ jobs: loadtest: runs-on: ubuntu-latest + permissions: + contents: read + packages: write steps: - name: Clone repository uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3 + - name: Login to GHCR + run: echo "${{ secrets.GITHUB_TOKEN }}" | docker login ghcr.io -u ${{ github.actor }} --password-stdin - name: Build loadtest - run: docker build -f ./build/package/loadtest.dockerfile . + run: docker build -f ./build/package/loadtest.dockerfile -t ghcr.io/hatchet-dev/hatchet/hatchet-loadtest:testing . + # Mutable "testing" tag for pulling into a staging k8s cluster without a + # release - every PR that touches this job overwrites it, so it always + # reflects whichever PR pushed most recently, not any one branch. + - name: Push loadtest (testing tag) + run: docker push ghcr.io/hatchet-dev/hatchet/hatchet-loadtest:testing loadtest-arm: runs-on: hatchet-arm64-2 diff --git a/cmd/hatchet-loadtest/timing.go b/cmd/hatchet-loadtest/timing.go index b0715adeed..e1628b9596 100644 --- a/cmd/hatchet-loadtest/timing.go +++ b/cmd/hatchet-loadtest/timing.go @@ -8,16 +8,29 @@ import ( "time" "github.com/google/uuid" + "golang.org/x/sync/errgroup" "github.com/hatchet-dev/hatchet/pkg/client/rest" v1 "github.com/hatchet-dev/hatchet/pkg/v1" //nolint:staticcheck // SA1019: used only for REST timing queries in --externalWorker mode ) -// timingSeenTTL bounds the memory used by TimingCollector.seen: entries -// older than this are pruned on each sweep, since a run id is only needed -// long enough to avoid re-fetching its (already-terminal) timings. +// timingSeenTTL bounds the memory used by TimingCollector.seen and +// TimingCollector.pending: successful entries are pruned since a run id is +// only needed long enough to avoid re-fetching its (already-terminal) +// timings, and pending entries are given up on (with a warning) after this +// long of repeated fetch failures, so a permanently-broken run id doesn't +// retry forever. const timingSeenTTL = 5 * time.Minute +// timingFetchConcurrency bounds how many V1WorkflowRunGetTimings calls the +// collector has in flight at once. Fetching one run's timings per REST +// round-trip is inherently an N+1 query pattern (there's no batch timings +// endpoint), so at load-test throughputs a single sequential fetch loop +// falls further behind every sweep and never catches up. Fetching +// concurrently, bounded by this limit, lets the collector actually drain +// the backlog instead of piling it up until it's abandoned at shutdown. +const timingFetchConcurrency = 50 + // timingPageLimit is the page size used when listing workflow runs. const timingPageLimit int64 = 100 @@ -104,7 +117,8 @@ func tryResolveWorkflowIDs(ctx context.Context, api *rest.ClientWithResponses, t type TimingCollector struct { lastSeen time.Time api *rest.ClientWithResponses - seen map[uuid.UUID]time.Time + seen map[uuid.UUID]time.Time // successfully fetched; value is fetch time + pending map[uuid.UUID]time.Time // discovered but not yet successfully fetched; value is first-discovered time workflowIds []uuid.UUID pollInterval time.Duration mu sync.Mutex @@ -122,6 +136,7 @@ func NewTimingCollector(hatchet v1.HatchetClient, workflowIds []uuid.UUID, pollI // up runs that finished just before the collector started. lastSeen: time.Now().Add(-pollInterval), seen: make(map[uuid.UUID]time.Time), + pending: make(map[uuid.UUID]time.Time), } } @@ -173,31 +188,32 @@ func (c *TimingCollector) sweep(ctx context.Context, out chan<- PhaseSample) { resp, err := c.api.V1WorkflowRunListWithResponse(ctx, c.tenantId, params) if err != nil { l.Warn().Err(err).Msg("timing collector: error listing workflow runs") - return + break } if resp.JSON200 == nil { - return + break } rows := resp.JSON200.Rows + c.mu.Lock() for _, row := range rows { runId := row.WorkflowRunExternalId - c.mu.Lock() - _, alreadySeen := c.seen[runId] - if !alreadySeen { - c.seen[runId] = now + // Already fetched, or already queued/in-flight from an earlier + // sweep (including one still being retried after a prior + // failure) - don't queue it again. + if _, ok := c.seen[runId]; ok { + continue } - c.mu.Unlock() - - if alreadySeen { + if _, ok := c.pending[runId]; ok { continue } - c.fetchTimings(ctx, runId, out) + c.pending[runId] = now } + c.mu.Unlock() if int64(len(rows)) < timingPageLimit { break @@ -206,6 +222,43 @@ func (c *TimingCollector) sweep(ctx context.Context, out chan<- PhaseSample) { offset += timingPageLimit } + // Fetch every currently-pending run - newly discovered above, plus any + // carried over from a previous sweep whose fetch failed - concurrently, + // bounded by timingFetchConcurrency. A run is only removed from + // `pending` (and added to `seen`) once its fetch actually succeeds, so a + // failed fetch (including one aborted by ctx cancellation on shutdown) + // just gets retried on the next sweep instead of silently and + // permanently dropping that run's samples. + c.mu.Lock() + toFetch := make([]uuid.UUID, 0, len(c.pending)) + for id := range c.pending { + toFetch = append(toFetch, id) + } + c.mu.Unlock() + + var wg errgroup.Group + wg.SetLimit(timingFetchConcurrency) + + for _, runId := range toFetch { + runId := runId + + wg.Go(func() error { + if err := c.fetchTimings(ctx, runId, out); err != nil { + l.Warn().Err(err).Str("workflow_run_id", runId.String()).Msg("timing collector: error fetching task timings") + return nil + } + + c.mu.Lock() + delete(c.pending, runId) + c.seen[runId] = time.Now() + c.mu.Unlock() + + return nil + }) + } + + _ = wg.Wait() // fetchTimings never returns a non-nil error to the group; failures are handled (and logged) above + c.mu.Lock() c.lastSeen = now for id, seenAt := range c.seen { @@ -213,20 +266,25 @@ func (c *TimingCollector) sweep(ctx context.Context, out chan<- PhaseSample) { delete(c.seen, id) } } + for id, firstSeen := range c.pending { + if now.Sub(firstSeen) > timingSeenTTL { + l.Warn().Str("workflow_run_id", id.String()).Msg("timing collector: giving up on workflow run after repeated fetch failures") + delete(c.pending, id) + } + } c.mu.Unlock() } -func (c *TimingCollector) fetchTimings(ctx context.Context, runId uuid.UUID, out chan<- PhaseSample) { +func (c *TimingCollector) fetchTimings(ctx context.Context, runId uuid.UUID, out chan<- PhaseSample) error { var depth int64 resp, err := c.api.V1WorkflowRunGetTimingsWithResponse(ctx, runId, &rest.V1WorkflowRunGetTimingsParams{Depth: &depth}) if err != nil { - l.Warn().Err(err).Str("workflow_run_id", runId.String()).Msg("timing collector: error fetching task timings") - return + return err } if resp.JSON200 == nil { - return + return fmt.Errorf("unexpected response: %s", resp.Status()) } for _, row := range resp.JSON200.Rows { @@ -245,7 +303,9 @@ func (c *TimingCollector) fetchTimings(ctx context.Context, runId uuid.UUID, out select { case out <- sample: case <-ctx.Done(): - return + return ctx.Err() } } + + return nil } From a6e9dd3983d5a25f8d41da5409c36a844a136241 Mon Sep 17 00:00:00 2001 From: Julius Park Date: Mon, 27 Jul 2026 15:48:40 -0400 Subject: [PATCH 2/5] try this --- cmd/hatchet-loadtest/do.go | 55 +++++++++++++++++++++++++++++----- cmd/hatchet-loadtest/timing.go | 15 ++++++++++ 2 files changed, 63 insertions(+), 7 deletions(-) diff --git a/cmd/hatchet-loadtest/do.go b/cmd/hatchet-loadtest/do.go index 66a549fc90..23bf9eecb6 100644 --- a/cmd/hatchet-loadtest/do.go +++ b/cmd/hatchet-loadtest/do.go @@ -256,15 +256,22 @@ func do(config LoadTestConfig) error { var phaseSamples chan PhaseSample var phaseResultCh <-chan phaseAccumulator var cancelTiming context.CancelFunc + var collector *TimingCollector if config.ExternalWorker { workflowIDs := <-resolvedWorkflowIDs + // Deliberately not derived from `ctx`: `ctx`'s deadline is sized for + // registration + emission + config.Wait, with no slack for however + // long the collector needs to drain its backlog below. Tying the + // collector to it would silently truncate that drain once the + // overall test deadline hit, which is exactly the kind of premature + // cutoff we're trying to avoid. var timingCtx context.Context - timingCtx, cancelTiming = context.WithCancel(ctx) + timingCtx, cancelTiming = context.WithCancel(context.Background()) defer cancelTiming() // safe to call more than once; guards every return path below - collector := NewTimingCollector(timingClient, workflowIDs, 2*time.Second) + collector = NewTimingCollector(timingClient, workflowIDs, timingPollInterval) phaseSamples = make(chan PhaseSample, 256) phaseResultCh = accumulatePhases(phaseSamples) @@ -286,16 +293,50 @@ func do(config LoadTestConfig) error { var phases phaseAccumulator if config.ExternalWorker { - // Give the engine config.Wait to finish (and the collector to observe) runs that - // were still in flight when emission stopped, before tearing down collection - - // cancelling immediately here would abort the collector mid-sweep for any run that - // completes right around this instant, silently dropping its timing sample instead - // of just not counting it (see the "context canceled" warnings this produces). + // Give the engine config.Wait to finish runs that were still in flight + // when emission stopped. if config.Wait > 0 { l.Info().Msgf("externalWorker: waiting %s for in-flight runs to complete before stopping timing collection...", config.Wait) time.Sleep(config.Wait) } + // Then keep the collector running until it has actually fetched + // every run it has discovered - cancelling on a fixed timer instead + // aborts whatever's still queued (however much that is) and + // permanently drops those samples. A run only ever leaves the + // collector's backlog by being successfully fetched or by aging out + // after timingSeenTTL (logged when that happens), so this loop is + // guaranteed to terminate on its own. + // + // Pending() hitting 0 isn't enough by itself: it can be a brief gap + // between sweeps, right before the next list pass turns up more + // newly-completed runs. Require it to stay at 0 across a full poll + // cycle (long enough for at least one more list pass to confirm + // there's really nothing left) before calling it drained. + l.Info().Msg("externalWorker: waiting for timing collector to fetch all discovered runs...") + logTicker := time.NewTicker(5 * time.Second) + var zeroSince time.Time + for { + pending := collector.Pending() + + if pending == 0 { + if zeroSince.IsZero() { + zeroSince = time.Now() + } else if time.Since(zeroSince) >= 2*timingPollInterval { + break + } + } else { + zeroSince = time.Time{} + } + + select { + case <-logTicker.C: + l.Info().Msgf("externalWorker: still waiting on %d run(s)...", pending) + case <-time.After(200 * time.Millisecond): + } + } + logTicker.Stop() + cancelTiming() phases = <-phaseResultCh } diff --git a/cmd/hatchet-loadtest/timing.go b/cmd/hatchet-loadtest/timing.go index e1628b9596..da269a69a6 100644 --- a/cmd/hatchet-loadtest/timing.go +++ b/cmd/hatchet-loadtest/timing.go @@ -31,6 +31,10 @@ const timingSeenTTL = 5 * time.Minute // the backlog instead of piling it up until it's abandoned at shutdown. const timingFetchConcurrency = 50 +// timingPollInterval is how often the collector re-lists for newly +// completed workflow runs. +const timingPollInterval = 2 * time.Second + // timingPageLimit is the page size used when listing workflow runs. const timingPageLimit int64 = 100 @@ -140,6 +144,17 @@ func NewTimingCollector(hatchet v1.HatchetClient, workflowIds []uuid.UUID, pollI } } +// Pending reports how many discovered runs are still awaiting a successful +// fetch (in flight or queued for retry). Callers that need every result - +// rather than whatever happened to be fetched before some fixed deadline - +// can poll this and only tear the collector down once it's been 0 for a +// full poll cycle (see do.go). +func (c *TimingCollector) Pending() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.pending) +} + // Run polls until ctx is done, sending a PhaseSample on out for every task // timing row with a full queued/scheduling/execution triple. func (c *TimingCollector) Run(ctx context.Context, out chan<- PhaseSample) { From 84c1f0368d55d7118208c2632db8431d36e4fd27 Mon Sep 17 00:00:00 2001 From: Julius Park Date: Mon, 27 Jul 2026 16:15:42 -0400 Subject: [PATCH 3/5] add sampling rate --- cmd/hatchet-loadtest/do.go | 23 ++++++++++++++++----- cmd/hatchet-loadtest/main.go | 8 ++++++++ cmd/hatchet-loadtest/timing.go | 37 +++++++++++++++++++++++++++------- 3 files changed, 56 insertions(+), 12 deletions(-) diff --git a/cmd/hatchet-loadtest/do.go b/cmd/hatchet-loadtest/do.go index 23bf9eecb6..54f5bca15b 100644 --- a/cmd/hatchet-loadtest/do.go +++ b/cmd/hatchet-loadtest/do.go @@ -271,7 +271,7 @@ func do(config LoadTestConfig) error { timingCtx, cancelTiming = context.WithCancel(context.Background()) defer cancelTiming() // safe to call more than once; guards every return path below - collector = NewTimingCollector(timingClient, workflowIDs, timingPollInterval) + collector = NewTimingCollector(timingClient, workflowIDs, timingPollInterval, config.TimingSampleRate) phaseSamples = make(chan PhaseSample, 256) phaseResultCh = accumulatePhases(phaseSamples) @@ -344,17 +344,30 @@ func do(config LoadTestConfig) error { expected := int64(config.EventFanout) * emitted * int64(config.DagSteps) if config.ExternalWorker { + sampleRate := int64(config.TimingSampleRate) + if sampleRate < 1 { + sampleRate = 1 + } + expectedSampled := expected / sampleRate + log.Printf( - "ℹ️ pushed %d, using %d events/s (externalWorker: engine-observed samples — queued n=%d, scheduling n=%d, execution n=%d)", - emitted, config.Events, phases.queued.count, phases.scheduling.count, phases.execution.count, + "ℹ️ pushed %d, using %d events/s (externalWorker: engine-observed samples, 1-in-%d sampled — queued n=%d, scheduling n=%d, execution n=%d)", + emitted, config.Events, sampleRate, phases.queued.count, phases.scheduling.count, phases.execution.count, ) if phases.execution.count == 0 { return fmt.Errorf("❌ no timing samples observed - check that the external SDK worker actually executed tasks for workflow(s) %v", expectedWorkflowNames(timingClient.V0().Namespace(), config.EventFanout)) } - if expected != phases.execution.count { - log.Printf("⚠️ warning: pushed and executed-timing-sample counts do not match: expected=%d got=%d", expected, phases.execution.count) + // Sampling only targets 1-in-sampleRate, not an exact count, and the + // collector already waits for a full drain (see the Pending() loop + // above) - so this is a sanity check that the sample landed in the + // right ballpark, not a completeness check. + if expectedSampled > 0 { + lower, upper := expectedSampled/2, expectedSampled+expectedSampled/2 + if phases.execution.count < lower || phases.execution.count > upper { + log.Printf("⚠️ warning: engine-observed sample count is well outside the expected range: expected≈%d (1-in-%d sample of %d pushed) got=%d", expectedSampled, sampleRate, expected, phases.execution.count) + } } } else { // NOTE: `emit()` returns successfully pushed events (not merely generated IDs), diff --git a/cmd/hatchet-loadtest/main.go b/cmd/hatchet-loadtest/main.go index a6935fe926..c71d2eac72 100644 --- a/cmd/hatchet-loadtest/main.go +++ b/cmd/hatchet-loadtest/main.go @@ -45,6 +45,13 @@ type LoadTestConfig struct { // compatible workflow named "load-test-0" (or "load-test-{i}" for // i Date: Tue, 28 Jul 2026 15:56:03 -0400 Subject: [PATCH 4/5] try it without batch task --- cmd/hatchet-loadtest/go/main.go | 34 ++++++++++++++++----------------- 1 file changed, 17 insertions(+), 17 deletions(-) diff --git a/cmd/hatchet-loadtest/go/main.go b/cmd/hatchet-loadtest/go/main.go index 5c8266f670..1e7ae67a0c 100644 --- a/cmd/hatchet-loadtest/go/main.go +++ b/cmd/hatchet-loadtest/go/main.go @@ -65,7 +65,7 @@ func run() error { delayMs := envInt("HATCHET_LOADTEST_DELAY_MS", 0) failureRate := envFloat("HATCHET_LOADTEST_FAILURE_RATE", 0) workerName := envOr("HATCHET_LOADTEST_WORKER_NAME", "load-test-worker") - batchTaskName := envOr("HATCHET_LOADTEST_BATCH_WORKFLOW_NAME", "load-test-batch") + // batchTaskName := envOr("HATCHET_LOADTEST_BATCH_WORKFLOW_NAME", "load-test-batch") task := client.NewStandaloneTask(taskName, func(ctx hatchet.Context, input LoadTestInput) (LoadTestOutput, error) { took := time.Since(input.CreatedAt) @@ -94,25 +94,25 @@ func run() error { // deliberately doesn't match the "load-test-%d" pattern that cmd/hatchet-loadtest's // expectedWorkflowNames() (do.go) resolves, so the benchmark's TimingCollector never // discovers or polls it and its timings never affect the pass/fail thresholds. - batchTask := client.NewStandaloneBatchTask(batchTaskName, func(ctx hatchet.Context, tasks map[string]LoadTestInput) (map[string]LoadTestOutput, error) { - out := make(map[string]LoadTestOutput, len(tasks)) - for id := range tasks { - out[id] = LoadTestOutput{ - Message: "This ran at: " + time.Now().Format(time.RFC3339Nano), - } - } - return out, nil - }, - hatchet.BatchConfig{ - MaxSize: 10, - MaxInterval: durationPtr(500 * time.Millisecond), - }, - hatchet.WithWorkflowEvents(eventKey), - ) + // batchTask := client.NewStandaloneBatchTask(batchTaskName, func(ctx hatchet.Context, tasks map[string]LoadTestInput) (map[string]LoadTestOutput, error) { + // out := make(map[string]LoadTestOutput, len(tasks)) + // for id := range tasks { + // out[id] = LoadTestOutput{ + // Message: "This ran at: " + time.Now().Format(time.RFC3339Nano), + // } + // } + // return out, nil + // }, + // hatchet.BatchConfig{ + // MaxSize: 10, + // MaxInterval: durationPtr(500 * time.Millisecond), + // }, + // hatchet.WithWorkflowEvents(eventKey), + //) worker, err := client.NewWorker( workerName, - hatchet.WithWorkflows(task, batchTask), + hatchet.WithWorkflows(task), ) if err != nil { return fmt.Errorf("failed to create worker: %w", err) From 0089e53e83b818378cc849e2f4306d1d536f23a7 Mon Sep 17 00:00:00 2001 From: Julius Park Date: Tue, 28 Jul 2026 16:26:39 -0400 Subject: [PATCH 5/5] blah --- .github/workflows/build.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 3001129206..b28263f85f 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -266,12 +266,12 @@ jobs: - name: Login to GHCR run: echo "${{ secrets.GITHUB_TOKEN }}" | docker login ghcr.io -u ${{ github.actor }} --password-stdin - name: Build loadtest - run: docker build -f ./build/package/loadtest.dockerfile -t ghcr.io/hatchet-dev/hatchet/hatchet-loadtest:testing . + run: docker build -f ./build/package/loadtest.dockerfile -t ghcr.io/hatchet-dev/hatchet/hatchet-loadtest:testing-no-batch . # Mutable "testing" tag for pulling into a staging k8s cluster without a # release - every PR that touches this job overwrites it, so it always # reflects whichever PR pushed most recently, not any one branch. - name: Push loadtest (testing tag) - run: docker push ghcr.io/hatchet-dev/hatchet/hatchet-loadtest:testing + run: docker push ghcr.io/hatchet-dev/hatchet/hatchet-loadtest:testing-no-batch loadtest-arm: runs-on: hatchet-arm64-2