Live throttle control on the run console

internal/engine/runner.go

@@ -81,6 +81,12 @@ func (r *Runner) WithProgress(fn ProgressFunc) *Runner {
 	return r
 }
 
+// SetRate changes the throttle rate (rows/sec, 0 = unlimited) of a running
+// runner, live. Used by the dashboard speed slider.
+func (r *Runner) SetRate(rowsPerSec int) {
+	r.bucket.SetRate(rowsPerSec)
+}
+
 // Run blocks until the run reaches a terminal state or ctx is canceled.
 func (r *Runner) Run(ctx context.Context) {
 	if err := r.run(ctx); err != nil {

internal/handler/run/handler.go

@@ -94,9 +94,11 @@ func (h *Handler) Get(ctx *gofr.Context) (any, error) {
 
 type controlRequest struct {
 	Action string `json:"action"`
+	Rate   *int64 `json:"rate"` // for action "speed": new rows/sec, 0 = unlimited
 }
 
-// Control handles PATCH /runs/{id} with {"action": "pause"|"resume"|"kill"}.
+// Control handles PATCH /runs/{id} with {"action": "pause"|"resume"|"kill"} or
+// {"action": "speed", "rate": N} to change the throttle live.
 func (h *Handler) Control(ctx *gofr.Context) (any, error) {
 	id, err := strconv.ParseInt(ctx.PathParam("id"), 10, 64)
 	if err != nil {
@@ -108,6 +110,19 @@ func (h *Handler) Control(ctx *gofr.Context) (any, error) {
 		return nil, gofrhttp.ErrorInvalidParam{Params: []string{"body"}}
 	}
 
+	if req.Action == "speed" {
+		if req.Rate == nil {
+			return nil, gofrhttp.ErrorInvalidParam{Params: []string{"rate"}}
+		}
+
+		run, err := h.orch.SetSpeed(ctx, id, *req.Rate)
+		if err != nil {
+			return nil, err
+		}
+
+		return toStatus(run), nil
+	}
+
 	run, err := h.orch.Control(ctx, id, req.Action)
 	if err != nil {
 		return nil, err

internal/service/orchestrator/service.go

@@ -38,9 +38,16 @@ var (
 	ErrUnknownAction = errors.New("unknown action (want pause, resume, or kill)")
 )
 
+// activeRun is a runner live in this process: its cancel func and the runner
+// itself (so we can adjust its throttle).
+type activeRun struct {
+	cancel context.CancelFunc
+	runner *engine.Runner
+}
+
 type Orchestrator struct {
 	mu     sync.Mutex
-	active map[int64]context.CancelFunc // runID → cancel
+	active map[int64]*activeRun // runID → live runner
 
 	hub *stream.Hub
 	rdb *redis.Client // non-nil enables fleet (lease) mode
@@ -54,7 +61,7 @@ type Orchestrator struct {
 // New builds a solo-mode orchestrator (control plane + embedded sequential runner).
 func New(hub *stream.Hub) *Orchestrator {
 	return &Orchestrator{
-		active: make(map[int64]context.CancelFunc),
+		active: make(map[int64]*activeRun),
 		hub:    hub,
 	}
 }
@@ -368,13 +375,43 @@ func (o *Orchestrator) ScheduleTick(ctx *gofr.Context) {
 	}
 }
 
+// SetSpeed changes a run's throttle rate live (rows/sec, 0 = unlimited). It
+// adjusts the running runner's token bucket immediately (solo mode) and
+// persists the rate on the task so it survives reconnects and reloads.
+func (o *Orchestrator) SetSpeed(ctx *gofr.Context, runID, rate int64) (models.Run, error) {
+	if rate < 0 {
+		rate = 0
+	}
+
+	run, err := o.runs.Get(ctx, ctx.SQL, runID)
+	if err != nil {
+		return models.Run{}, err
+	}
+
+	if err := o.tasks.SetRate(ctx, ctx.SQL, run.TaskID, int(rate)); err != nil {
+		return models.Run{}, err
+	}
+
+	o.mu.Lock()
+	if a, ok := o.active[runID]; ok && a.runner != nil {
+		a.runner.SetRate(int(rate))
+	}
+	o.mu.Unlock()
+
+	if err := o.events.Append(ctx, ctx.SQL, runID, "api", "run.speed", fmt.Sprintf("%d rows/sec", rate)); err != nil {
+		ctx.Logger.Errorf("run %d: audit write failed: %v", runID, err)
+	}
+
+	return o.runs.Get(ctx, ctx.SQL, runID)
+}
+
 // Shutdown cancels all live runners (used on process exit).
 func (o *Orchestrator) Shutdown() {
 	o.mu.Lock()
 	defer o.mu.Unlock()
 
-	for _, cancel := range o.active {
-		cancel()
+	for _, a := range o.active {
+		a.cancel()
 	}
 }
 
@@ -387,13 +424,14 @@ func (o *Orchestrator) spawn(c *container.Container, task models.Task, runID int
 	}
 
 	runCtx, cancel := context.WithCancel(context.Background())
-	o.active[runID] = cancel
 
 	runner := engine.NewRunner(c.SQL, c.Logger, task, runID)
 	if o.hub != nil {
 		runner.WithProgress(o.hub.Publish)
 	}
 
+	o.active[runID] = &activeRun{cancel: cancel, runner: runner}
+
 	go func() {
 		defer func() {
 			o.mu.Lock()
@@ -419,7 +457,7 @@ func (o *Orchestrator) cancel(runID int64) {
 	o.mu.Lock()
 	defer o.mu.Unlock()
 
-	if cancel, ok := o.active[runID]; ok {
-		cancel()
+	if a, ok := o.active[runID]; ok {
+		a.cancel()
 	}
 }

internal/store/task.go

@@ -28,6 +28,16 @@ func (TaskStore) Create(ctx context.Context, db DB, t *models.Task) error {
 	return nil
 }
 
+// SetRate updates a task's throttle rate (rows/sec, 0 = unlimited).
+func (TaskStore) SetRate(ctx context.Context, db DB, id int64, rate int) error {
+	_, err := db.ExecContext(ctx, `UPDATE tasks SET rate_per_sec = $1 WHERE id = $2`, rate, id)
+	if err != nil {
+		return fmt.Errorf("set task %d rate: %w", id, err)
+	}
+
+	return nil
+}
+
 func (TaskStore) Get(ctx context.Context, db DB, id int64) (models.Task, error) {
 	var t models.Task
 

web/index.html

@@ -5,9 +5,10 @@
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <title>MARATHON — data operations control</title>
     <script>
-      // Apply saved theme before paint to avoid a flash.
+      // Apply theme before paint to avoid a flash (URL ?theme= wins, then saved).
       try {
-        document.documentElement.dataset.theme = localStorage.getItem("marathon-theme") || "light";
+        var p = new URLSearchParams(location.search).get("theme");
+        document.documentElement.dataset.theme = p || localStorage.getItem("marathon-theme") || "light";
       } catch (e) {}
     </script>
   </head>

web/src/App.tsx

@@ -19,7 +19,10 @@ function initialView(): View {
 
 export function App() {
   const [view, setView] = useState<View>(initialView);
-  const [theme, setTheme] = useState<Theme>(() => (localStorage.getItem("marathon-theme") as Theme) || "light");
+  const [theme, setTheme] = useState<Theme>(() => {
+    const p = new URLSearchParams(location.search).get("theme");
+    return (p as Theme) || (localStorage.getItem("marathon-theme") as Theme) || "light";
+  });
 
   useEffect(() => {
     document.documentElement.dataset.theme = theme;

web/src/api.ts

@@ -38,6 +38,11 @@ export const api = {
       method: "PATCH",
       body: JSON.stringify({ action }),
     }),
+  setSpeed: (id: number, rate: number) =>
+    call<Run>(`/runs/${id}`, {
+      method: "PATCH",
+      body: JSON.stringify({ action: "speed", rate }),
+    }),
   events: (id: number) => call<RunEvent[]>(`/runs/${id}/events`),
   quarantine: (id: number) =>
     call<QuarantinedRow[]>(`/runs/${id}/quarantine`),

web/src/pages/RunLive.tsx

@@ -19,7 +19,30 @@ export function RunLive({ runId, onBack }: { runId: number; onBack: () => void }
   const [quarantine, setQuarantine] = useState<QuarantinedRow[]>([]);
   const [events, setEvents] = useState<RunEvent[]>([]);
   const [notice, setNotice] = useState("");
+  const [speed, setSpeed] = useState<number | null>(null);
+  const [speedMax, setSpeedMax] = useState(100000);
   const last = useRef<{ t: number; processed: number } | null>(null);
+  const speedTimer = useRef<ReturnType<typeof setTimeout> | undefined>(undefined);
+
+  // Load the task's current throttle to initialize the slider.
+  useEffect(() => {
+    if (!run) return;
+    api.getTask(run.task_id)
+      .then((t) => {
+        setSpeed(t.rate_per_sec);
+        setSpeedMax(Math.max(100000, (t.rate_per_sec || 0) * 2));
+      })
+      .catch(() => {});
+    // eslint-disable-next-line react-hooks/exhaustive-deps
+  }, [run?.task_id]);
+
+  const onSpeed = (v: number) => {
+    setSpeed(v);
+    if (speedTimer.current) clearTimeout(speedTimer.current);
+    speedTimer.current = setTimeout(() => {
+      api.setSpeed(runId, v).catch((e) => setNotice((e as Error).message));
+    }, 250);
+  };
 
   useEffect(() => {
     api.getRun(runId).then(setRun).catch(() => {});
@@ -116,6 +139,24 @@ export function RunLive({ runId, onBack }: { runId: number; onBack: () => void }
               <AreaChart data={history} />
             </div>
 
+            {active && speed !== null && (
+              <div className="speed">
+                <div className="cap">
+                  <span className="l">Throttle</span>
+                  <span className="r tnum">{speed === 0 ? "Max (unthrottled)" : `${fmt(speed)} rows/s`}</span>
+                </div>
+                <input
+                  type="range"
+                  min={0}
+                  max={speedMax}
+                  step={500}
+                  value={speed}
+                  onChange={(e) => onSpeed(Number(e.target.value))}
+                  aria-label="Throttle rate"
+                />
+              </div>
+            )}
+
             <div className="controls">
               {active && (
                 <>

web/src/styles.css

@@ -269,6 +269,24 @@ a { color: var(--accent); text-decoration: none; }
 .chart { display: block; width: 100%; color: var(--accent); }
 .chart .grid { stroke: var(--grid); }
 
+.speed { margin: 16px 0 4px; }
+.speed .cap { display: flex; justify-content: space-between; margin-bottom: 8px; }
+.speed .cap .l { font-size: 10.5px; text-transform: uppercase; letter-spacing: 1.1px; color: var(--faint); font-weight: 600; }
+.speed .cap .r { font-family: var(--mono); font-size: 12px; color: var(--ink-2); }
+.speed input[type="range"] {
+  -webkit-appearance: none; appearance: none; width: 100%; height: 6px; border-radius: 999px;
+  background: var(--paper-2); outline: none; cursor: pointer;
+}
+.speed input[type="range"]::-webkit-slider-thumb {
+  -webkit-appearance: none; appearance: none; width: 18px; height: 18px; border-radius: 50%;
+  background: var(--accent); border: 3px solid var(--card); box-shadow: 0 1px 4px #0003; cursor: pointer;
+}
+.speed input[type="range"]::-moz-range-thumb {
+  width: 18px; height: 18px; border-radius: 50%; background: var(--accent);
+  border: 3px solid var(--card); box-shadow: 0 1px 4px #0003; cursor: pointer;
+}
+.speed input[type="range"]:focus-visible { box-shadow: 0 0 0 3px var(--accent-ring); }
+
 .controls { display: flex; gap: 10px; margin-top: 20px; flex-wrap: wrap; align-items: center; }
 .done-note { color: var(--good); font-weight: 550; font-size: 13.5px; }