Emil_Shanaty/SciMesh
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove the demo control room; /ui lands on the admin console
coordinator / test (push) Canceled after 0s
Details
python / test (push) Canceled after 0s
Details
release / binaries (amd64, darwin) (push) Canceled after 0s
Details
release / binaries (amd64, linux) (push) Canceled after 0s
Details
release / binaries (amd64, windows) (push) Canceled after 0s
Details
release / binaries (arm64, darwin) (push) Canceled after 0s
Details
release / binaries (arm64, linux) (push) Canceled after 0s
Details
release / binaries (arm64, windows) (push) Canceled after 0s
Details
release / image (push) Canceled after 0s
Details
users / test (push) Canceled after 0s
Details
release / release (push) Canceled after 0s
Details

Browse Source
This commit is contained in:
Emil
2026-08-03 01:58:02 +03:00
parent c4d88c7ffc
commit 12499d2d7b
24 changed files with 65 additions and 491 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coordinator/internal/storage/postgres/integration_test.go
+4 -4
View File
@@ -145,20 +145,20 @@ func TestUIReadRepoListsReducerFields(t *testing.T) {
if claimed, err := jobs.ClaimReduction(ctx, job.ID, time.Now().UTC()); err != nil || !claimed {
t.Fatalf("claim reduction = (%v, %v)", claimed, err)
}
listed, err := NewUIReadRepo(pool).ListJobs(ctx, nil, 20)
listed, _, err := NewAdminReadRepo(pool).ListJobsPaginated(ctx, "", 20, 0)
if err != nil {
t.Fatalf("list UI jobs: %v", err)
t.Fatalf("list admin jobs: %v", err)
}
for _, item := range listed {
if item.ID != job.ID {
continue
}
if item.Status != domain.JobReducing || item.ReducerStartedAt == nil {
t.Fatalf("UI reducer projection = %+v", item)
t.Fatalf("admin reducer projection = %+v", item)
}
return
}
t.Fatalf("seeded job %s is missing from UI list", job.ID)
t.Fatalf("seeded job %s is missing from the admin list", job.ID)
}
// A job must land whole or not at all: a half-created job leaves chunks no
coordinator/internal/storage/postgres/ui_read_repo.go
+1 -85
View File
@@ -24,40 +24,6 @@ func (r *UIReadRepo) GetJob(ctx context.Context, id uuid.UUID) (*domain.Job, err
return job, err
}
func (r *UIReadRepo) ListJobs(ctx context.Context, owner *uuid.UUID, limit int) ([]domain.Job, error) {
if limit < 1 || limit > 100 {
return nil, domain.ErrInvalidInput
}
q := psql.Select(jobColumns...).From("jobs")
if owner != nil {
q = q.Where(sq.Eq{"owner_id": *owner})
}
sql, args, err := q.OrderBy("created_at DESC", "id DESC").Limit(uint64(limit)).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 jobs: %w", err)
}
defer rows.Close()
jobs := make([]domain.Job, 0)
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()
}
func (r *UIReadRepo) ListTasksByJob(ctx context.Context, jobID uuid.UUID) ([]domain.Task, error) {
sql, args, err := psql.Select(taskColumns...).From("tasks").Where(sq.Eq{"job_id": jobID}).OrderBy("chunk_index ASC").ToSql()
if err != nil {
@@ -79,68 +45,18 @@ func (r *UIReadRepo) ListTasksByJob(ctx context.Context, jobID uuid.UUID) ([]dom
return tasks, rows.Err()
}
func (r *UIReadRepo) ListTasksByJobs(ctx context.Context, jobIDs []uuid.UUID) (map[uuid.UUID][]domain.Task, error) {
out := make(map[uuid.UUID][]domain.Task, len(jobIDs))
if len(jobIDs) == 0 {
return out, nil
}
sql, args, err := psql.Select(taskColumns...).From("tasks").
Where(sq.Eq{"job_id": jobIDs}).OrderBy("job_id ASC", "chunk_index 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 tasks by jobs: %w", err)
}
defer rows.Close()
for rows.Next() {
task, err := scanTask(rows)
if err != nil {
return nil, err
}
out[task.JobID] = append(out[task.JobID], *task)
}
return out, rows.Err()
}
func (r *UIReadRepo) ListWorkers(ctx context.Context, limit int) ([]domain.Worker, error) {
if limit < 1 || limit > 100 {
return nil, domain.ErrInvalidInput
}
sql, args, err := psql.Select(workerColumns...).From("workers").OrderBy("last_heartbeat_at DESC", "id DESC").Limit(uint64(limit)).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 workers: %w", err)
}
defer rows.Close()
workers := make([]domain.Worker, 0)
for rows.Next() {
worker, err := scanWorker(rows)
if err != nil {
return nil, err
}
workers = append(workers, *worker)
}
return workers, rows.Err()
}
func (r *UIReadRepo) ListWorkersByOwner(ctx context.Context, owner uuid.UUID, limit int) ([]domain.Worker, error) {
if limit < 1 || limit > 100 {
return nil, domain.ErrInvalidInput
}
sql, args, err := psql.Select(workerColumns...).From("workers").
Where(sq.Eq{"owner_id": owner}).
OrderBy("last_heartbeat_at DESC", "id DESC").Limit(uint64(limit)).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 workers by owner: %w", err)
return nil, fmt.Errorf("list workers: %w", err)
}
defer rows.Close()
workers := make([]domain.Worker, 0)
coordinator/internal/storage/sqlite/ui_read_repo.go
+2 -69
View File
@@ -4,7 +4,6 @@ import (
"context"
"database/sql"
"fmt"
"strings"
"github.com/google/uuid"
@@ -20,34 +19,6 @@ func (r *UIReadRepo) GetJob(ctx context.Context, id uuid.UUID) (*domain.Job, err
return NewJobRepo(r.db).Get(ctx, id)
}
func (r *UIReadRepo) ListJobs(ctx context.Context, owner *uuid.UUID, limit int) ([]domain.Job, error) {
if limit < 1 || limit > 100 {
return nil, domain.ErrInvalidInput
}
query := "SELECT " + jobColumns + " FROM jobs"
args := []any{}
if owner != nil {
query += " WHERE owner_id = ?"
args = append(args, owner.String())
}
query += " ORDER BY created_at DESC, id DESC LIMIT ?"
args = append(args, limit)
rows, err := conn(ctx, r.db).QueryContext(ctx, query, args...)
if err != nil {
return nil, fmt.Errorf("list jobs: %w", err)
}
defer func() { _ = rows.Close() }()
jobs := make([]domain.Job, 0)
for rows.Next() {
job, err := scanJob(rows)
if err != nil {
return nil, err
}
jobs = append(jobs, *job)
}
return jobs, rows.Err()
}
func (r *UIReadRepo) ListTasksByJob(ctx context.Context, jobID uuid.UUID) ([]domain.Task, error) {
rows, err := conn(ctx, r.db).QueryContext(ctx,
"SELECT "+taskColumns+" FROM tasks WHERE job_id = ? ORDER BY chunk_index ASC", jobID.String())
@@ -66,50 +37,12 @@ func (r *UIReadRepo) ListTasksByJob(ctx context.Context, jobID uuid.UUID) ([]dom
return tasks, rows.Err()
}
func (r *UIReadRepo) ListTasksByJobs(ctx context.Context, jobIDs []uuid.UUID) (map[uuid.UUID][]domain.Task, error) {
out := make(map[uuid.UUID][]domain.Task, len(jobIDs))
if len(jobIDs) == 0 {
return out, nil
}
placeholders := make([]string, 0, len(jobIDs))
args := make([]any, 0, len(jobIDs))
for _, id := range jobIDs {
placeholders = append(placeholders, "?")
args = append(args, id.String())
}
rows, err := conn(ctx, r.db).QueryContext(ctx,
"SELECT "+taskColumns+" FROM tasks WHERE job_id IN ("+strings.Join(placeholders, ", ")+") ORDER BY job_id ASC, chunk_index ASC",
args...)
if err != nil {
return nil, fmt.Errorf("list tasks by jobs: %w", err)
}
defer func() { _ = rows.Close() }()
for rows.Next() {
task, err := scanTask(rows)
if err != nil {
return nil, err
}
out[task.JobID] = append(out[task.JobID], *task)
}
return out, rows.Err()
}
func (r *UIReadRepo) ListWorkers(ctx context.Context, limit int) ([]domain.Worker, error) {
return r.listWorkers(ctx, "", nil, limit)
}
func (r *UIReadRepo) ListWorkersByOwner(ctx context.Context, owner uuid.UUID, limit int) ([]domain.Worker, error) {
return r.listWorkers(ctx, " WHERE owner_id = ?", []any{owner.String()}, limit)
}
func (r *UIReadRepo) listWorkers(ctx context.Context, clause string, args []any, limit int) ([]domain.Worker, error) {
if limit < 1 || limit > 100 {
return nil, domain.ErrInvalidInput
}
query := "SELECT " + workerColumns + " FROM workers" + clause +
" ORDER BY last_heartbeat_at DESC, id DESC LIMIT ?"
fullArgs := append(args, limit)
rows, err := conn(ctx, r.db).QueryContext(ctx, query, fullArgs...)
rows, err := conn(ctx, r.db).QueryContext(ctx,
"SELECT "+workerColumns+" FROM workers ORDER BY last_heartbeat_at DESC, id DESC LIMIT ?", limit)
if err != nil {
return nil, fmt.Errorf("list workers: %w", err)
}