diff --git a/coordinator/cmd/coordinator/main.go b/coordinator/cmd/coordinator/main.go index 90d352f..9b84204 100644 --- a/coordinator/cmd/coordinator/main.go +++ b/coordinator/cmd/coordinator/main.go @@ -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{ diff --git a/coordinator/internal/agent/daemon.go b/coordinator/internal/agent/daemon.go index 3af610f..1143fbf 100644 --- a/coordinator/internal/agent/daemon.go +++ b/coordinator/internal/agent/daemon.go @@ -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 { diff --git a/coordinator/internal/agent/setupui/server.go b/coordinator/internal/agent/setupui/server.go index cafacdd..8927076 100644 --- a/coordinator/internal/agent/setupui/server.go +++ b/coordinator/internal/agent/setupui/server.go @@ -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 { diff --git a/coordinator/internal/agent/setupui/server_test.go b/coordinator/internal/agent/setupui/server_test.go index f86b020..d76d524 100644 --- a/coordinator/internal/agent/setupui/server_test.go +++ b/coordinator/internal/agent/setupui/server_test.go @@ -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) + } +} diff --git a/coordinator/internal/agent/setupui/template.html b/coordinator/internal/agent/setupui/template.html index 43c8c73..dd91808 100644 --- a/coordinator/internal/agent/setupui/template.html +++ b/coordinator/internal/agent/setupui/template.html @@ -174,6 +174,12 @@ code{font-family:var(--mono);font-size:.86em}

Worker is working

—
+
+
—claimed
+
—completed
+
—failed
+
—registered
+
—
@@ -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=''+v.coordinator;meta.append(c)} diff --git a/coordinator/internal/memstore/memstore.go b/coordinator/internal/memstore/memstore.go index ace397f..f0e6881 100644 --- a/coordinator/internal/memstore/memstore.go +++ b/coordinator/internal/memstore/memstore.go @@ -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 +} diff --git a/coordinator/internal/storage/postgres/job_repo.go b/coordinator/internal/storage/postgres/job_repo.go index eb859f2..6b2a9b1 100644 --- a/coordinator/internal/storage/postgres/job_repo.go +++ b/coordinator/internal/storage/postgres/job_repo.go @@ -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 +} diff --git a/coordinator/internal/storage/postgres/worker_repo.go b/coordinator/internal/storage/postgres/worker_repo.go index e3d2bc5..92e58a3 100644 --- a/coordinator/internal/storage/postgres/worker_repo.go +++ b/coordinator/internal/storage/postgres/worker_repo.go @@ -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 +} diff --git a/coordinator/internal/storage/sqlite/admin_m2_test.go b/coordinator/internal/storage/sqlite/admin_m2_test.go index 46f5c3d..99a7f04 100644 --- a/coordinator/internal/storage/sqlite/admin_m2_test.go +++ b/coordinator/internal/storage/sqlite/admin_m2_test.go @@ -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") + } +} diff --git a/coordinator/internal/storage/sqlite/job_repo.go b/coordinator/internal/storage/sqlite/job_repo.go index 1616a54..98974ec 100644 --- a/coordinator/internal/storage/sqlite/job_repo.go +++ b/coordinator/internal/storage/sqlite/job_repo.go @@ -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 +} diff --git a/coordinator/internal/storage/sqlite/worker_repo.go b/coordinator/internal/storage/sqlite/worker_repo.go index 3dbfc17..207de6e 100644 --- a/coordinator/internal/storage/sqlite/worker_repo.go +++ b/coordinator/internal/storage/sqlite/worker_repo.go @@ -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 +} diff --git a/coordinator/internal/transport/http/server.go b/coordinator/internal/transport/http/server.go index 1bc6e95..98fb9af 100644 --- a/coordinator/internal/transport/http/server.go +++ b/coordinator/internal/transport/http/server.go @@ -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) diff --git a/coordinator/internal/transport/http/templates/admin.html b/coordinator/internal/transport/http/templates/admin.html index 9f2dfb4..253f04b 100644 --- a/coordinator/internal/transport/http/templates/admin.html +++ b/coordinator/internal/transport/http/templates/admin.html @@ -210,7 +210,7 @@ tbody tr:hover{background:var(--panel-2)}
Workers register themselves. Trust decides whether a machine's results are accepted directly or need quorum.
- +
WorkerStatusCapabilitiesTrustOwnerLast signal
WorkerStatusCapabilitiesTrustOwnerLast signal
@@ -284,6 +284,12 @@ tbody tr:hover{background:var(--panel-2)}
The cluster token below authenticates any worker. Reveal it only on a trusted machine.
+
Danger zone
+
+
+
Prune artifacts
Delete finished jobs older than a cutoff and all their artifacts (blob files included).
+
+
Cluster
@@ -440,9 +446,15 @@ async function loadWorkers(){ ''+(w.capabilities||[]).map(c=>''+esc(c)+'').join('')+''+ ''+trustSel+''+ ''+esc(w.owner)+''+ - ''+fmtTime(w.last_heartbeat_at)+''; + ''+fmtTime(w.last_heartbeat_at)+''+ + ''+(w.status==='offline'?'':'')+''; 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(); +}); diff --git a/coordinator/internal/transport/http/ui_admin.go b/coordinator/internal/transport/http/ui_admin.go index 47f3bad..dc3e31f 100644 --- a/coordinator/internal/transport/http/ui_admin.go +++ b/coordinator/internal/transport/http/ui_admin.go @@ -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) +} diff --git a/coordinator/internal/usecase/admin.go b/coordinator/internal/usecase/admin.go index 0e2a233..bd3812b 100644 --- a/coordinator/internal/usecase/admin.go +++ b/coordinator/internal/usecase/admin.go @@ -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) +} diff --git a/coordinator/internal/usecase/admin_test.go b/coordinator/internal/usecase/admin_test.go index 19ff436..6a8b64b 100644 --- a/coordinator/internal/usecase/admin_test.go +++ b/coordinator/internal/usecase/admin_test.go @@ -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") + } +} diff --git a/coordinator/internal/usecase/ports.go b/coordinator/internal/usecase/ports.go index 1798529..00a5f48 100644 --- a/coordinator/internal/usecase/ports.go +++ b/coordinator/internal/usecase/ports.go @@ -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; diff --git a/coordinator/internal/usecase/prune.go b/coordinator/internal/usecase/prune.go new file mode 100644 index 0000000..4a8a076 --- /dev/null +++ b/coordinator/internal/usecase/prune.go @@ -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 +} diff --git a/coordinator/internal/usecase/usecase_test.go b/coordinator/internal/usecase/usecase_test.go index b1e29c3..4819b65 100644 --- a/coordinator/internal/usecase/usecase_test.go +++ b/coordinator/internal/usecase/usecase_test.go @@ -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") + } +}