diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 081342acc8..b28263f85f 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-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-no-batch
 
   loadtest-arm:
     runs-on: hatchet-arm64-2
diff --git a/cmd/hatchet-loadtest/do.go b/cmd/hatchet-loadtest/do.go
index 66a549fc90..54f5bca15b 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, config.TimingSampleRate)
 
 		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
 	}
@@ -303,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/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)
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<EventFanout).
 	ExternalWorker bool
+
+	// TimingSampleRate, in --externalWorker mode, only fetches full task
+	// timings for 1 out of every TimingSampleRate completed runs, rather
+	// than every one - the reported average converges from a uniform sample
+	// just as well as from the full set, at a fraction of the REST load on
+	// the engine. 1 (or less) fetches every run.
+	TimingSampleRate int
 }
 
 func main() {
@@ -98,6 +105,7 @@ func main() {
 	loadtest.Flags().DurationVar(&config.AverageDurationThreshold, "averageDurationThreshold", 100*time.Millisecond, "averageDurationThreshold specifies the threshold for the average duration per executed event to be considered a success")
 	loadtest.Flags().StringVar(&config.PlotDir, "plotDirectory", "", "plotDirectory specifies where to put the generated plots for latency and task duration")
 	loadtest.Flags().BoolVar(&config.ExternalWorker, "externalWorker", false, "externalWorker skips registering a workflow and starting an in-process worker, assuming a separately-running SDK worker (e.g. cmd/hatchet-loadtest/go) has already registered a compatible workflow; worker/workflow flags (slots, dagSteps, eventFanout, rlKeys, workerDelay, failureRate, delay) are ignored in this mode")
+	loadtest.Flags().IntVar(&config.TimingSampleRate, "timingSampleRate", 10, "in --externalWorker mode, fetch full task timings for 1 out of every N completed runs instead of every one - the average still converges, at a fraction of the REST load on the engine; 1 fetches every run")
 	cmd := &cobra.Command{Use: "app"}
 	cmd.AddCommand(loadtest)
 	if err := cmd.Execute(); err != nil {
diff --git a/cmd/hatchet-loadtest/timing.go b/cmd/hatchet-loadtest/timing.go
index b0715adeed..43d89c261a 100644
--- a/cmd/hatchet-loadtest/timing.go
+++ b/cmd/hatchet-loadtest/timing.go
@@ -8,16 +8,33 @@ 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
+
+// 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
 
@@ -104,27 +121,51 @@ 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, or deliberately skipped by sampling; value is decision time
+	pending      map[uuid.UUID]time.Time // sampled-in and awaiting a successful fetch; value is first-discovered time
 	workflowIds  []uuid.UUID
 	pollInterval time.Duration
+	sampleRate   int   // fetch full timings for 1 out of every sampleRate discovered runs; 1 = every run
+	discovered   int64 // count of runs discovered so far, used to pick the 1-in-sampleRate subset
 	mu           sync.Mutex
 	tenantId     uuid.UUID
 }
 
 // NewTimingCollector builds a collector for already-resolved workflow ids.
-func NewTimingCollector(hatchet v1.HatchetClient, workflowIds []uuid.UUID, pollInterval time.Duration) *TimingCollector { //nolint:staticcheck // SA1019
+// sampleRate < 2 fetches full timings for every discovered run; sampleRate
+// of N fetches roughly 1 out of every N - an average over a uniform sample
+// converges to the same value much faster than one over every run, at a
+// fraction of the REST load on the engine.
+func NewTimingCollector(hatchet v1.HatchetClient, workflowIds []uuid.UUID, pollInterval time.Duration, sampleRate int) *TimingCollector { //nolint:staticcheck // SA1019
+	if sampleRate < 1 {
+		sampleRate = 1
+	}
+
 	return &TimingCollector{
 		api:          hatchet.V0().API(),
 		tenantId:     uuid.MustParse(hatchet.V0().TenantId()),
 		workflowIds:  workflowIds,
 		pollInterval: pollInterval,
+		sampleRate:   sampleRate,
 		// Start the window slightly in the past so the first sweep can pick
 		// 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),
 	}
 }
 
+// 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) {
@@ -173,31 +214,44 @@ 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 skipped by sampling), or already
+			// queued/in-flight from an earlier sweep (including one still
+			// being retried after a prior failure) - don't reconsider it.
+			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.discovered++
+
+			// Keep 1 out of every sampleRate discovered runs - always
+			// including the first, so a short test still gets at least one
+			// sample instead of a spurious "no timing samples observed"
+			// error. Runs that aren't selected are marked seen immediately
+			// (not fetched, never retried) so they don't keep costing a
+			// List-window reconsideration every sweep.
+			if (c.discovered-1)%int64(c.sampleRate) == 0 {
+				c.pending[runId] = now
+			} else {
+				c.seen[runId] = now
+			}
 		}
+		c.mu.Unlock()
 
 		if int64(len(rows)) < timingPageLimit {
 			break
@@ -206,6 +260,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 +304,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 +341,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
 }