diff --git a/docs/screenshots/run-dark.png b/docs/screenshots/run-dark.png new file mode 100644 index 0000000..d8f2afc Binary files /dev/null and b/docs/screenshots/run-dark.png differ diff --git a/docs/screenshots/run-light.png b/docs/screenshots/run-light.png new file mode 100644 index 0000000..b970276 Binary files /dev/null and b/docs/screenshots/run-light.png differ diff --git a/docs/screenshots/tasks.png b/docs/screenshots/tasks.png new file mode 100644 index 0000000..0bac92b Binary files /dev/null and b/docs/screenshots/tasks.png differ diff --git a/internal/engine/runner.go b/internal/engine/runner.go index 49c7fb3..97022b6 100644 --- a/internal/engine/runner.go +++ b/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 { diff --git a/internal/handler/run/handler.go b/internal/handler/run/handler.go index 78f6d7c..35cc836 100644 --- a/internal/handler/run/handler.go +++ b/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 diff --git a/internal/service/orchestrator/service.go b/internal/service/orchestrator/service.go index dea8993..2501e83 100644 --- a/internal/service/orchestrator/service.go +++ b/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() } } diff --git a/internal/store/task.go b/internal/store/task.go index 5e37578..c5aad6e 100644 --- a/internal/store/task.go +++ b/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 diff --git a/web/index.html b/web/index.html index c8852c8..73c3532 100644 --- a/web/index.html +++ b/web/index.html @@ -5,9 +5,10 @@ MARATHON — data operations control diff --git a/web/src/App.tsx b/web/src/App.tsx index cc6bdc9..014a663 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -19,7 +19,10 @@ function initialView(): View { export function App() { const [view, setView] = useState(initialView); - const [theme, setTheme] = useState(() => (localStorage.getItem("marathon-theme") as Theme) || "light"); + const [theme, setTheme] = useState(() => { + 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; diff --git a/web/src/api.ts b/web/src/api.ts index 4298d2e..f6e3243 100644 --- a/web/src/api.ts +++ b/web/src/api.ts @@ -38,6 +38,11 @@ export const api = { method: "PATCH", body: JSON.stringify({ action }), }), + setSpeed: (id: number, rate: number) => + call(`/runs/${id}`, { + method: "PATCH", + body: JSON.stringify({ action: "speed", rate }), + }), events: (id: number) => call(`/runs/${id}/events`), quarantine: (id: number) => call(`/runs/${id}/quarantine`), diff --git a/web/src/pages/RunLive.tsx b/web/src/pages/RunLive.tsx index 1ff8516..b6fce88 100644 --- a/web/src/pages/RunLive.tsx +++ b/web/src/pages/RunLive.tsx @@ -19,7 +19,30 @@ export function RunLive({ runId, onBack }: { runId: number; onBack: () => void } const [quarantine, setQuarantine] = useState([]); const [events, setEvents] = useState([]); const [notice, setNotice] = useState(""); + const [speed, setSpeed] = useState(null); + const [speedMax, setSpeedMax] = useState(100000); const last = useRef<{ t: number; processed: number } | null>(null); + const speedTimer = useRef | 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 } + {active && speed !== null && ( +
+
+ Throttle + {speed === 0 ? "Max (unthrottled)" : `${fmt(speed)} rows/s`} +
+ onSpeed(Number(e.target.value))} + aria-label="Throttle rate" + /> +
+ )} +
{active && ( <> diff --git a/web/src/styles.css b/web/src/styles.css index 1fb3f88..912c0f7 100644 --- a/web/src/styles.css +++ b/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; }