Add worker stats to the wizard, artifact pruning and offline-worker removal to the admin console

This commit is contained in:
Emil
2026-08-03 04:11:43 +03:00
parent 7ea7c52325
commit 357ed34714
19 changed files with 465 additions and 12 deletions
+1
View File
@@ -173,6 +173,7 @@ func runWithConfig(cfg infra.Config) error {
GetJobResult: usecase.NewGetJobResult(jobRepo, usecase.NewDownloadArtifact(artifactRepo, blobStore)),
GetTaskInput: usecase.NewGetTaskInput(taskRepo, artifactRepo, blobStore),
Dashboard: usecase.NewDashboard(uiReadRepo, catalog),
PruneArtifacts: usecase.NewPruneArtifacts(jobRepo, uiReadRepo, blobStore, clk),
PreviewArtifact: usecase.NewPreviewArtifact(uiReadRepo, blobStore),
Admin: usecase.NewAdmin(deps.adminReadRepo, uiReadRepo, workerRepo, deps.settingsRepo, catalog,
usecase.AdminNodeInfo{
+1
View File
@@ -158,6 +158,7 @@ func (d *Daemon) runOnce() (Outcome, error) {
if task == nil {
return Outcome{Claimed: false}, nil
}
d.log.Info("task claimed", "task_id", task.TaskID, "attempt", task.Attempt)
started := time.Now()
taskDir := filepath.Join(d.config.WorkDir, task.TaskID, fmt.Sprint(task.Attempt))
if err := os.MkdirAll(taskDir, 0o750); err != nil {
+43 -10
View File
@@ -289,15 +289,26 @@ func writeJSON(w http.ResponseWriter, status int, v any) {
// statusView is what the wizard needs to paint the running/stopped state.
type statusView struct {
ConfigPresent bool `json:"config_present"`
ConfigPath string `json:"config_path"`
LogPath string `json:"log_path"`
Running bool `json:"running"`
Pid int `json:"pid"`
WorkerName string `json:"worker_name,omitempty"`
Coordinator string `json:"coordinator,omitempty"`
WorkDir string `json:"work_dir,omitempty"`
TokenSet bool `json:"token_set"`
ConfigPresent bool `json:"config_present"`
ConfigPath string `json:"config_path"`
LogPath string `json:"log_path"`
Running bool `json:"running"`
Pid int `json:"pid"`
WorkerName string `json:"worker_name,omitempty"`
Coordinator string `json:"coordinator,omitempty"`
WorkDir string `json:"work_dir,omitempty"`
TokenSet bool `json:"token_set"`
Stats WorkerStats `json:"stats"`
}
// WorkerStats is parsed from the worker log: the agent reports each claim,
// completion and failure as a structured line, so the wizard can show live
// counters without any coordinator access.
type WorkerStats struct {
Registered bool `json:"registered"`
Claimed int `json:"claimed"`
Completed int `json:"completed"`
Failed int `json:"failed"`
}
// ensureVenvTaskRunner rewrites the saved config so its task runner uses the
@@ -319,8 +330,30 @@ func (s *Server) ensureVenvTaskRunner() {
}
}
// parseWorkerStats counts the structured agent events in the worker log.
func parseWorkerStats(logPath string) WorkerStats {
var stats WorkerStats
raw, err := os.ReadFile(logPath)
if err != nil {
return stats
}
for _, line := range strings.Split(string(raw), "\n") {
switch {
case strings.Contains(line, "msg=registered"):
stats.Registered = true
case strings.Contains(line, `msg="task claimed"`):
stats.Claimed++
case strings.Contains(line, `msg="task completed"`):
stats.Completed++
case strings.Contains(line, `msg="task failed"`):
stats.Failed++
}
}
return stats
}
func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) {
view := statusView{ConfigPath: s.cfgPath, LogPath: s.logPath, Running: s.sup.Alive(), Pid: s.sup.Pid()}
view := statusView{ConfigPath: s.cfgPath, LogPath: s.logPath, Running: s.sup.Alive(), Pid: s.sup.Pid(), Stats: parseWorkerStats(s.logPath)}
if raw, err := os.ReadFile(s.cfgPath); err == nil {
var file agent.ConfigFile
if json.Unmarshal(raw, &file) == nil {
@@ -479,3 +479,21 @@ func TestTestProbesTheVenvPythonAfterInstall(t *testing.T) {
t.Errorf("scimesh check = %+v, want the venv interpreter reporting 9.9.9-test", report.Scimesh)
}
}
func TestParseWorkerStats(t *testing.T) {
logPath := filepath.Join(t.TempDir(), "worker.log")
content := `time=1 level=INFO msg=registered worker_id=w1
time=2 level=INFO msg="task claimed" task_id=t1 attempt=0
time=3 level=INFO msg="task completed" task_id=t1 elapsed_seconds=2
time=4 level=INFO msg="task claimed" task_id=t2 attempt=0
time=5 level=WARN msg="task failed" task_id=t2 error_code=X retryable=true
time=6 level=WARN msg="agent cycle failed" error="boom"
`
if err := os.WriteFile(logPath, []byte(content), 0o600); err != nil {
t.Fatal(err)
}
stats := parseWorkerStats(logPath)
if !stats.Registered || stats.Claimed != 2 || stats.Completed != 1 || stats.Failed != 1 {
t.Errorf("stats = %+v, want registered claimed=2 completed=1 failed=1", stats)
}
}
@@ -174,6 +174,12 @@ code{font-family:var(--mono);font-size:.86em}
<div><h1 id="st-title">Worker is working</h1><div class="sub" id="st-sub">—</div></div>
<button class="btn btn-danger" id="st-stop" style="margin-left:auto"><svg viewBox="0 0 24 24" fill="none" stroke-width="2"><rect x="6" y="6" width="12" height="12" rx="2"/></svg>Stop</button>
</div>
<div class="stat-row">
<div class="stat"><b id="st-claimed">—</b><span>claimed</span></div>
<div class="stat"><b id="st-completed">—</b><span>completed</span></div>
<div class="stat bad"><b id="st-failed">—</b><span>failed</span></div>
<div class="stat"><b id="st-registered">—</b><span>registered</span></div>
</div>
<div class="meta-line" id="st-meta"></div>
<div class="logbox" id="st-log"></div>
<div class="actions"><span class="link" id="st-cfg">—</span><button class="btn btn-ghost" id="st-reconfig">Reconfigure…</button></div>
@@ -293,6 +299,11 @@ async function refreshStatus(){
$('st-title').textContent=v.running?(v.worker_name||'Worker')+' is working':(v.worker_name||'Worker')+' is stopped';
$('st-sub').textContent='pid '+(v.pid||'—')+' · config '+(v.config_present?v.config_path:'not saved yet');
$('st-cfg').textContent='Configuration: '+(v.config_present?v.config_path:'—');
const stats=v.stats||{};
$('st-claimed').textContent=stats.claimed!=null?stats.claimed:'—';
$('st-completed').textContent=stats.completed!=null?stats.completed:'—';
$('st-failed').textContent=stats.failed!=null?stats.failed:'—';
$('st-registered').textContent=stats.registered?'yes':'no';
const meta=$('st-meta');
meta.innerHTML='';
if(v.coordinator){const c=document.createElement('span');c.className='chip';c.innerHTML='<svg viewBox="0 0 24 24" fill="none" stroke-width="2"><circle cx="12" cy="12" r="9"/></svg>'+v.coordinator;meta.append(c)}
+29
View File
@@ -461,3 +461,32 @@ func (r *TaskResultRepo) CountAgreeing(_ context.Context, taskID uuid.UUID, sha2
}
return n, nil
}
func (r *JobRepo) ListCompletedBefore(ctx context.Context, cutoff time.Time) ([]domain.Job, error) {
r.mu.Lock()
defer r.mu.Unlock()
var out []domain.Job
for _, j := range r.jobs {
if j.CompletedAt != nil && j.CompletedAt.Before(cutoff) {
out = append(out, *j)
}
}
return out, nil
}
func (r *JobRepo) Delete(ctx context.Context, id uuid.UUID) error {
r.mu.Lock()
defer r.mu.Unlock()
delete(r.jobs, id)
return nil
}
func (r *WorkerRepo) Delete(ctx context.Context, id uuid.UUID) error {
r.mu.Lock()
defer r.mu.Unlock()
if _, ok := r.workers[id]; !ok {
return domain.ErrWorkerNotFound
}
delete(r.workers, id)
return nil
}
@@ -3,6 +3,7 @@ package postgres
import (
"context"
"errors"
"fmt"
"time"
sq "github.com/Masterminds/squirrel"
@@ -153,3 +154,46 @@ func (r *JobRepo) UpdateStatus(ctx context.Context, id uuid.UUID,
}
return nil
}
// ListCompletedBefore returns jobs whose completion timestamp is older than
// the cutoff (completed and failed both count as finished).
func (r *JobRepo) ListCompletedBefore(ctx context.Context, cutoff time.Time) ([]domain.Job, error) {
sql, args, err := psql.Select(jobColumns...).From("jobs").
Where(sq.NotEq{"completed_at": nil}).
Where(sq.Lt{"completed_at": cutoff}).
OrderBy("completed_at ASC").
ToSql()
if err != nil {
return nil, err
}
rows, err := conn(ctx, r.pool).Query(ctx, sql, args...)
if err != nil {
return nil, fmt.Errorf("list completed jobs: %w", err)
}
defer rows.Close()
var jobs []domain.Job
for rows.Next() {
var j domain.Job
var status string
if err := rows.Scan(
&j.ID, &j.Workload, &j.InputURI, &j.Parameters, &status, &j.CreatedAt, &j.CompletedAt,
&j.InputArtifactID, &j.ResultArtifactID, &j.ErrorCode, &j.ErrorMessage, &j.ReducerStartedAt,
&j.OwnerID,
); err != nil {
return nil, err
}
j.Status = domain.JobStatus(status)
jobs = append(jobs, j)
}
return jobs, rows.Err()
}
// Delete removes the job row; tasks, artifacts and task_results cascade.
func (r *JobRepo) Delete(ctx context.Context, id uuid.UUID) error {
sql, args, err := psql.Delete("jobs").Where(sq.Eq{"id": id}).ToSql()
if err != nil {
return err
}
_, err = conn(ctx, r.pool).Exec(ctx, sql, args...)
return err
}
@@ -123,3 +123,19 @@ func scanWorker(row pgx.Row) (*domain.Worker, error) {
w.TrustLevel = domain.WorkerTrust(trust)
return &w, nil
}
// Delete removes a worker from the registry.
func (r *WorkerRepo) Delete(ctx context.Context, id uuid.UUID) error {
sql, args, err := psql.Delete("workers").Where(sq.Eq{"id": id}).ToSql()
if err != nil {
return err
}
tag, err := conn(ctx, r.pool).Exec(ctx, sql, args...)
if err != nil {
return fmt.Errorf("delete worker: %w", err)
}
if tag.RowsAffected() == 0 {
return domain.ErrWorkerNotFound
}
return nil
}
@@ -88,3 +88,47 @@ func TestWorkerSetTrust(t *testing.T) {
t.Errorf("unknown worker trust err = %v, want ErrWorkerNotFound", err)
}
}
func TestJobRepoListCompletedBeforeAndDelete(t *testing.T) {
db := newTestDB(t)
ctx := context.Background()
repo := NewJobRepo(db)
old := seedJob(t, db, 2)
oldTime := fixedTime().Add(-40 * 24 * time.Hour)
if err := repo.UpdateStatus(ctx, old.ID, domain.JobCompleted, &oldTime); err != nil {
t.Fatal(err)
}
fresh := seedJob(t, db, 2)
freshTime := fixedTime().Add(-2 * time.Hour)
if err := repo.UpdateStatus(ctx, fresh.ID, domain.JobCompleted, &freshTime); err != nil {
t.Fatal(err)
}
// The failing check constraint needs no result artifact for completed; the
// UpdateStatus path is fine, but tasks stay pending — irrelevant here.
list, err := repo.ListCompletedBefore(ctx, fixedTime().Add(-7*24*time.Hour))
if err != nil {
t.Fatal(err)
}
if len(list) != 1 || list[0].ID != old.ID {
t.Errorf("list = %d jobs, want only the old one", len(list))
}
if err := repo.Delete(ctx, old.ID); err != nil {
t.Fatal(err)
}
var n int
if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM jobs WHERE id = ?", old.ID.String()).Scan(&n); err != nil {
t.Fatal(err)
}
if n != 0 {
t.Error("job row must be gone after Delete")
}
// Tasks cascaded away with the job.
if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM tasks WHERE job_id = ?", old.ID.String()).Scan(&n); err != nil {
t.Fatal(err)
}
if n != 0 {
t.Error("tasks must cascade with the job")
}
}
@@ -3,6 +3,7 @@ package sqlite
import (
"context"
"database/sql"
"fmt"
"time"
"github.com/google/uuid"
@@ -166,3 +167,30 @@ func nullableUUID(id *uuid.UUID) any {
}
return id.String()
}
// ListCompletedBefore returns jobs whose completion timestamp is older than
// the cutoff (completed and failed both count as finished).
func (r *JobRepo) ListCompletedBefore(ctx context.Context, cutoff time.Time) ([]domain.Job, error) {
rows, err := conn(ctx, r.db).QueryContext(ctx,
"SELECT "+jobColumns+" FROM jobs WHERE completed_at IS NOT NULL AND completed_at < ? ORDER BY completed_at ASC",
encodeTime(cutoff))
if err != nil {
return nil, fmt.Errorf("list completed jobs: %w", err)
}
defer func() { _ = rows.Close() }()
var jobs []domain.Job
for rows.Next() {
job, err := scanJob(rows)
if err != nil {
return nil, err
}
jobs = append(jobs, *job)
}
return jobs, rows.Err()
}
// Delete removes the job row; tasks, artifacts and task_results cascade.
func (r *JobRepo) Delete(ctx context.Context, id uuid.UUID) error {
_, err := conn(ctx, r.db).ExecContext(ctx, "DELETE FROM jobs WHERE id = ?", id.String())
return err
}
@@ -107,3 +107,19 @@ func (r *WorkerRepo) SetTrust(ctx context.Context, id uuid.UUID, trust domain.Wo
}
return nil
}
// Delete removes a worker from the registry.
func (r *WorkerRepo) Delete(ctx context.Context, id uuid.UUID) error {
res, err := conn(ctx, r.db).ExecContext(ctx, "DELETE FROM workers WHERE id = ?", id.String())
if err != nil {
return err
}
affected, err := res.RowsAffected()
if err != nil {
return err
}
if affected == 0 {
return domain.ErrWorkerNotFound
}
return nil
}
@@ -36,6 +36,7 @@ type UseCases struct {
Dashboard *usecase.Dashboard
PreviewArtifact *usecase.PreviewArtifact
Admin *usecase.Admin
PruneArtifacts *usecase.PruneArtifacts
}
type Server struct {
@@ -184,6 +185,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
ui.Handle("GET /ui/admin/api/metrics", chain(http.HandlerFunc(s.handleUIAdminMetricsJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/workers", chain(http.HandlerFunc(s.handleUIAdminWorkersJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/workers/{id}/trust", chain(http.HandlerFunc(s.handleUIAdminSetTrustJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/workers/{id}/remove", chain(http.HandlerFunc(s.handleUIAdminRemoveWorkerJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/users", chain(http.HandlerFunc(s.handleUIAdminUsersJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/users/{id}/role", chain(http.HandlerFunc(s.handleUIAdminSetUserRoleJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/worker-keys", chain(http.HandlerFunc(s.handleUIAdminWorkerKeysJSON), gate, requireAdmin))
@@ -192,6 +194,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
ui.Handle("POST /ui/admin/api/workloads/{name}/enabled", chain(http.HandlerFunc(s.handleUIAdminSetWorkloadEnabledJSON), gate, requireAdmin))
ui.Handle("GET /ui/admin/api/settings", chain(http.HandlerFunc(s.handleUIAdminSettingsJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/token/reveal", chain(http.HandlerFunc(s.handleUIAdminRevealTokenJSON), gate, requireAdmin))
ui.Handle("POST /ui/admin/api/prune", chain(http.HandlerFunc(s.handleUIAdminPruneJSON), gate, requireAdmin))
} else {
for _, rt := range app {
ui.HandleFunc(rt.pattern, rt.handler)
@@ -210,7 +210,7 @@ tbody tr:hover{background:var(--panel-2)}
<div class="section-note">Workers register themselves. Trust decides whether a machine's results are accepted directly or need quorum.</div>
<div class="card">
<table>
<thead><tr><th>Worker</th><th>Status</th><th>Capabilities</th><th>Trust</th><th>Owner</th><th>Last signal</th></tr></thead>
<thead><tr><th>Worker</th><th>Status</th><th>Capabilities</th><th>Trust</th><th>Owner</th><th>Last signal</th><th></th></tr></thead>
<tbody id="worker-rows"></tbody>
</table>
</div>
@@ -284,6 +284,12 @@ tbody tr:hover{background:var(--panel-2)}
<!-- ═══ SETTINGS ═══ -->
<section class="page" id="page-settings">
<div class="warn-strip" style="display:flex;gap:10px;align-items:flex-start;background:var(--amber-soft);border:1px solid #e5b64f33;border-radius:10px;padding:12px 14px;font-size:12.5px;color:#eecf8d"><span>The cluster token below authenticates <b>any</b> worker. Reveal it only on a trusted machine.</span></div>
<div class="section-title" style="color:var(--red)">Danger zone</div>
<div class="card" style="border-color:#f2647c33">
<dl class="kv">
<dt>Prune artifacts</dt><dd style="display:flex;justify-content:space-between;align-items:center;gap:14px"><span style="color:var(--text-2);font-size:12.5px">Delete finished jobs older than a cutoff and all their artifacts (blob files included).</span><span style="display:flex;gap:8px;align-items:center"><input id="prune-days" type="number" min="1" max="3650" value="30" style="width:80px"><button class="btn btn-danger btn-sm" id="prune">Prune…</button></span></dd>
</dl>
</div>
<div class="section-title">Cluster</div>
<div class="card">
<dl class="kv">
@@ -440,9 +446,15 @@ async function loadWorkers(){
'<td>'+(w.capabilities||[]).map(c=>'<span class="cap">'+esc(c)+'</span>').join('')+'</td>'+
'<td>'+trustSel+'</td>'+
'<td style="color:var(--text-2)">'+esc(w.owner)+'</td>'+
'<td style="color:var(--text-2);font-size:12.5px">'+fmtTime(w.last_heartbeat_at)+'</td>';
'<td style="color:var(--text-2);font-size:12.5px">'+fmtTime(w.last_heartbeat_at)+'</td>'+
'<td>'+(w.status==='offline'?'<button class="btn btn-danger btn-sm worker-remove" data-id="'+w.id+'" data-name="'+esc(w.name)+'">Remove</button>':'')+'</td>';
rows.append(tr);
}
document.querySelectorAll('.worker-remove').forEach(btn=>btn.addEventListener('click',async()=>{
if(!confirm('Remove the offline worker "'+btn.dataset.name+'" from the registry? This cannot be undone.'))return;
await fetch('/ui/admin/api/workers/'+btn.dataset.id+'/remove',{method:'POST'});
loadWorkers();
}));
document.querySelectorAll('.trust-sel').forEach(sel=>sel.addEventListener('change',async()=>{
await fetch('/ui/admin/api/workers/'+sel.dataset.id+'/trust',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({trusted:sel.value==='trusted'})});
loadWorkers();
@@ -530,6 +542,17 @@ document.getElementById('reveal').addEventListener('click',async e=>{
}else{tok.textContent='••••••••••••••••••••••••';e.target.textContent='Reveal'}
});
document.querySelectorAll('.toggle').forEach(t=>t.addEventListener('click',()=>t.classList.toggle('on')));
document.getElementById('prune').addEventListener('click',async()=>{
const days=parseInt(document.getElementById('prune-days').value||'30',10);
if(!confirm('Delete finished jobs older than '+days+' days with all their artifacts? This cannot be undone.'))return;
const btn=document.getElementById('prune');btn.disabled=true;btn.textContent='Pruning…';
const r=await fetch('/ui/admin/api/prune',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({older_than_days:days})});
btn.disabled=false;btn.textContent='Prune…';
if(!r.ok){alert('Prune failed.');return}
const v=await r.json();
alert('Removed '+v.jobs+' jobs and '+v.artifacts+' artifacts, freed '+fmtBytes(v.freed_bytes)+'.');
if(current.page==='settings')loadSettings();
});
</script>
</body>
</html>
@@ -8,6 +8,7 @@ import (
"net/url"
"strconv"
"strings"
"time"
"github.com/google/uuid"
@@ -390,3 +391,43 @@ func (s *Server) adminOwnerEmails(r *http.Request) map[uuid.UUID]string {
}
return out
}
// handleUIAdminPruneJSON deletes finished jobs older than the requested
// number of days (with all their artifacts) and reports what was freed.
func (s *Server) handleUIAdminPruneJSON(w http.ResponseWriter, r *http.Request) {
var body struct {
OlderThanDays int `json:"older_than_days"`
}
if err := decodeJSON(r, &body); err != nil {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
if body.OlderThanDays < 1 || body.OlderThanDays > 3650 {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
ctx, cancel := s.reqCtx(r)
defer cancel()
result, err := s.uc.PruneArtifacts.Execute(ctx, time.Duration(body.OlderThanDays)*24*time.Hour)
if err != nil {
s.writeError(w, r, err)
return
}
writeJSON(w, http.StatusOK, result)
}
// handleUIAdminRemoveWorkerJSON deletes an offline worker.
func (s *Server) handleUIAdminRemoveWorkerJSON(w http.ResponseWriter, r *http.Request) {
id, err := uuid.Parse(r.PathValue("id"))
if err != nil {
s.writeError(w, r, domain.ErrInvalidInput)
return
}
ctx, cancel := s.reqCtx(r)
defer cancel()
if err := s.uc.Admin.RemoveWorker(ctx, id); err != nil {
s.writeError(w, r, err)
return
}
w.WriteHeader(http.StatusNoContent)
}
+17
View File
@@ -521,3 +521,20 @@ func (a *Admin) RevealWorkerToken(ctx context.Context, actor string) string {
}
return token
}
// RemoveWorker deletes an offline worker from the registry. Online or busy
// workers are refused: an admin console must never yank a live machine out
// from under a running task.
func (a *Admin) RemoveWorker(ctx context.Context, id uuid.UUID) error {
if a.workers == nil {
return domain.ErrWorkerNotFound
}
worker, err := a.workers.Get(ctx, id)
if err != nil {
return err
}
if worker.Status != domain.WorkerOffline {
return domain.ErrInvalidInput
}
return a.workers.Delete(ctx, id)
}
@@ -336,3 +336,30 @@ func TestAdminRevealToken(t *testing.T) {
t.Errorf("token = %q", got)
}
}
type removableWorkerRepo struct {
WorkerRepository
deleted uuid.UUID
}
func (f *removableWorkerRepo) Get(ctx context.Context, id uuid.UUID) (*domain.Worker, error) {
return &domain.Worker{ID: id, Status: domain.WorkerOffline}, nil
}
func (f *removableWorkerRepo) Delete(ctx context.Context, id uuid.UUID) error {
f.deleted = id
return nil
}
func TestAdminRemoveWorker(t *testing.T) {
a := adminFixture()
repo := &removableWorkerRepo{}
a.workers = repo
id := uuid.New()
if err := a.RemoveWorker(context.Background(), id); err != nil {
t.Fatal(err)
}
if repo.deleted != id {
t.Error("offline worker must be deleted")
}
}
+9
View File
@@ -82,6 +82,12 @@ type JobRepository interface {
ClaimReduction(ctx context.Context, id uuid.UUID, startedAt time.Time) (bool, error)
CompleteWithResult(ctx context.Context, id, resultArtifactID uuid.UUID, completedAt time.Time) error
FailReduction(ctx context.Context, id uuid.UUID, code, message string, completedAt time.Time) error
// ListCompletedBefore returns jobs that finished (completed or failed)
// before the cutoff, for the admin artifact pruner.
ListCompletedBefore(ctx context.Context, cutoff time.Time) ([]domain.Job, error)
// Delete removes a job row; the engine cascades its tasks, artifacts and
// quorum votes. Blob files must be removed separately.
Delete(ctx context.Context, id uuid.UUID) error
}
// WorkerRepository persists the worker registry.
@@ -97,6 +103,9 @@ type WorkerRepository interface {
// SetTrust reclassifies a worker's trust level (trusted/untrusted). Returns
// ErrNotFound when the id is unknown.
SetTrust(ctx context.Context, id uuid.UUID, trust domain.WorkerTrust) error
// Delete removes a worker from the registry. Returns ErrNotFound when the
// id is unknown.
Delete(ctx context.Context, id uuid.UUID) error
}
// ArtifactRepository persists artifact metadata. The bytes live in a BlobStore;
+55
View File
@@ -0,0 +1,55 @@
package usecase
import (
"context"
"time"
)
// PruneArtifacts removes completed or failed jobs older than `olderThan` and
// every artifact they own: database rows cascade, blob files are deleted
// explicitly. It returns what was freed so the admin console can report it.
type PruneArtifacts struct {
jobs JobRepository
read UIReadRepository
blobs BlobStore
clk Clock
}
func NewPruneArtifacts(jobs JobRepository, read UIReadRepository, blobs BlobStore, clk Clock) *PruneArtifacts {
return &PruneArtifacts{jobs: jobs, read: read, blobs: blobs, clk: clk}
}
type PruneResult struct {
Jobs int `json:"jobs"`
Artifacts int `json:"artifacts"`
FreedBytes int64 `json:"freed_bytes"`
}
// Execute deletes finished jobs whose completion timestamp is older than the
// cutoff. Jobs that are still active are never touched.
func (uc *PruneArtifacts) Execute(ctx context.Context, olderThan time.Duration) (PruneResult, error) {
cutoff := uc.clk.Now().Add(-olderThan)
jobs, err := uc.jobs.ListCompletedBefore(ctx, cutoff)
if err != nil {
return PruneResult{}, err
}
out := PruneResult{}
for _, job := range jobs {
artifacts, err := uc.read.ListArtifactsByJob(ctx, job.ID)
if err != nil {
return out, err
}
for _, artifact := range artifacts {
if err := uc.blobs.Delete(ctx, artifact.StorageKey); err != nil {
return out, err
}
out.FreedBytes += artifact.SizeBytes
out.Artifacts++
}
if err := uc.jobs.Delete(ctx, job.ID); err != nil {
return out, err
}
out.Jobs++
}
return out, nil
}
@@ -980,3 +980,40 @@ func TestSubmitDatasetRejectsDisabledWorkload(t *testing.T) {
t.Fatalf("submit after re-enable: %v", err)
}
}
func TestPruneArtifactsRemovesOldFinishedJobs(t *testing.T) {
h := newHarness()
old := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC)
job := &domain.Job{ID: uuid.New(), Workload: "similarity-search", Status: domain.JobCompleted, CreatedAt: old, CompletedAt: &old}
if err := h.jobs.Insert(context.Background(), job); err != nil {
t.Fatal(err)
}
art, err := domain.NewArtifact(job.ID, nil, domain.ArtifactFinalResult, "r.csv", "text/csv", old)
if err != nil {
t.Fatal(err)
}
art.SetContent("sha", 42)
if err := h.arts.Insert(context.Background(), art); err != nil {
t.Fatal(err)
}
// An active job must survive the prune.
active := &domain.Job{ID: uuid.New(), Workload: "similarity-search", Status: domain.JobRunning, CreatedAt: old}
if err := h.jobs.Insert(context.Background(), active); err != nil {
t.Fatal(err)
}
prune := usecase.NewPruneArtifacts(h.jobs, memstore.NewUIReadRepo(h.jobs, h.tasks, h.work, h.arts), h.blobs, h.clk)
result, err := prune.Execute(context.Background(), 7*24*time.Hour)
if err != nil {
t.Fatal(err)
}
if result.Jobs != 1 || result.Artifacts != 1 || result.FreedBytes != 42 {
t.Errorf("prune = %+v, want 1 job / 1 artifact / 42 bytes", result)
}
if _, err := h.jobs.Get(context.Background(), job.ID); err == nil {
t.Error("finished job must be gone")
}
if _, err := h.jobs.Get(context.Background(), active.ID); err != nil {
t.Error("active job must survive the prune")
}
}