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

Drive coordinator upload and reduction from the workload catalog

Browse Source
This commit is contained in:
Emil
2026-08-02 17:47:47 +03:00
parent 700a96a259
commit 749396da05
26 changed files with 1473 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coordinator/internal/usecase/job.go
+9 -6
View File
@@ -7,6 +7,7 @@ import (
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
"github.com/emil28092005/SciMesh/coordinator/internal/workloads"
)
// Job operations: the submitter-facing lifecycle of a whole submission.
@@ -227,7 +228,7 @@ func progressFrom(job domain.Job, counts map[domain.TaskStatus]int) domain.JobPr
// Shared by CompleteTask and FailTask so both close a job by the same rule —
// the rule itself lives in domain.JobProgress.DeriveStatus.
func syncJobStatus(ctx context.Context, jobs JobRepository, tasks TaskRepository,
jobID uuid.UUID, now time.Time) error {
catalog *workloads.Catalog, jobID uuid.UUID, now time.Time) error {
counts, err := tasks.CountByStatus(ctx, jobID)
if err != nil {
@@ -239,9 +240,11 @@ func syncJobStatus(ctx context.Context, jobs JobRepository, tasks TaskRepository
}
status := progressFrom(*job, counts).DeriveStatus()
// All worker shards being complete means scientific reduction is ready, not
// that the job's final artifact already exists. CTX-09 owns the transition
// from reducing to completed after it persists that artifact.
if status == domain.JobCompleted && job.Workload == "similarity-search" {
// that the job's final artifact already exists. Known catalog workloads
// transition to reducing so ReduceJob can produce the final artifact; the
// reducer then completes the job with the result. Unknown (URI-based) jobs
// complete without a coordinator-owned final artifact.
if status == domain.JobCompleted && catalog != nil && catalog.Reduction(job.Workload) != "" {
status = domain.JobReducing
}
@@ -253,14 +256,14 @@ func syncJobStatus(ctx context.Context, jobs JobRepository, tasks TaskRepository
}
func syncExpiredJobStatuses(ctx context.Context, jobs JobRepository, tasks TaskRepository,
jobIDs []uuid.UUID, now time.Time) error {
catalog *workloads.Catalog, jobIDs []uuid.UUID, now time.Time) error {
seen := make(map[uuid.UUID]struct{}, len(jobIDs))
for _, jobID := range jobIDs {
if _, duplicate := seen[jobID]; duplicate {
continue
}
seen[jobID] = struct{}{}
if err := syncJobStatus(ctx, jobs, tasks, jobID, now); err != nil {
if err := syncJobStatus(ctx, jobs, tasks, catalog, jobID, now); err != nil {
return err
}
}
coordinator/internal/usecase/reduce.go
+22 -5
View File
@@ -9,6 +9,7 @@ import (
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
"github.com/emil28092005/SciMesh/coordinator/internal/reducer"
"github.com/emil28092005/SciMesh/coordinator/internal/workloads"
)
// ReduceJob turns completed coordinator-owned partial artifacts into one final
@@ -20,11 +21,12 @@ type ReduceJob struct {
blobs BlobStore
tx TxManager
clock Clock
catalog *workloads.Catalog
}
func NewReduceJob(jobs JobRepository, tasks TaskRepository, artifacts ArtifactRepository,
blobs BlobStore, tx TxManager, clock Clock) *ReduceJob {
return &ReduceJob{jobs: jobs, tasks: tasks, artifacts: artifacts, blobs: blobs, tx: tx, clock: clock}
blobs BlobStore, tx TxManager, clock Clock, catalog *workloads.Catalog) *ReduceJob {
return &ReduceJob{jobs: jobs, tasks: tasks, artifacts: artifacts, blobs: blobs, tx: tx, clock: clock, catalog: catalog}
}
// Execute is idempotent for jobs that are not currently reducing. The worker
@@ -42,7 +44,11 @@ func (uc *ReduceJob) Execute(ctx context.Context, jobID uuid.UUID) error {
if job.Status != domain.JobReducing {
return nil
}
if job.Workload != "similarity-search" {
if uc.catalog == nil {
return uc.fail(ctx, jobID)
}
reduction := uc.catalog.Reduction(job.Workload)
if reduction == "" {
return uc.fail(ctx, jobID)
}
@@ -73,13 +79,13 @@ func (uc *ReduceJob) Execute(ctx context.Context, jobID uuid.UUID) error {
readers = append(readers, body)
closers = append(closers, body)
}
output, reduceErr := reducer.ReduceSimilaritySearch(readers, job.Parameters)
output, reduceErr := reducePartials(reduction, readers, job.Parameters)
closeAll(closers)
if reduceErr != nil {
return uc.fail(ctx, jobID)
}
final, err := domain.NewArtifact(jobID, nil, domain.ArtifactFinalResult, "similarity-search.csv", "text/csv", uc.clock.Now())
final, err := domain.NewArtifact(jobID, nil, domain.ArtifactFinalResult, job.Workload+".csv", "text/csv", uc.clock.Now())
if err != nil {
return uc.fail(ctx, jobID)
}
@@ -100,6 +106,17 @@ func (uc *ReduceJob) Execute(ctx context.Context, jobID uuid.UUID) error {
return nil
}
func reducePartials(reduction string, readers []io.Reader, parameters map[string]any) ([]byte, error) {
switch reduction {
case "top-k":
return reducer.ReduceSimilaritySearch(readers, parameters)
case "ordered-concat":
return reducer.ReduceOrderedConcat(readers)
default:
return nil, domain.ErrInvalidInput
}
}
func (uc *ReduceJob) fail(ctx context.Context, jobID uuid.UUID) error {
// The public state carries a stable sanitized failure, never parser/storage
// internals that may include local paths or implementation details.
coordinator/internal/usecase/task.go
+22 -17
View File
@@ -8,6 +8,7 @@ import (
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
"github.com/emil28092005/SciMesh/coordinator/internal/workloads"
)
// Task operations: the worker-facing lifecycle of a single chunk.
@@ -27,10 +28,11 @@ type ClaimTask struct {
tx TxManager
clock Clock
leaseDuration time.Duration
catalog *workloads.Catalog
}
func NewClaimTask(tasks TaskRepository, jobs JobRepository, workers WorkerRepository, tx TxManager, clock Clock, leaseDuration time.Duration) *ClaimTask {
return &ClaimTask{tasks: tasks, jobs: jobs, workers: workers, tx: tx, clock: clock, leaseDuration: leaseDuration}
func NewClaimTask(tasks TaskRepository, jobs JobRepository, workers WorkerRepository, tx TxManager, clock Clock, leaseDuration time.Duration, catalog *workloads.Catalog) *ClaimTask {
return &ClaimTask{tasks: tasks, jobs: jobs, workers: workers, tx: tx, clock: clock, leaseDuration: leaseDuration, catalog: catalog}
}
// Execute reclaims elapsed leases first, then hands out one task.
@@ -77,7 +79,7 @@ func (uc *ClaimTask) Execute(ctx context.Context, in ClaimTaskInput) (*domain.Cl
if err != nil {
return err
}
if err := syncExpiredJobStatuses(ctx, uc.jobs, uc.tasks, affectedJobs, now); err != nil {
if err := syncExpiredJobStatuses(ctx, uc.jobs, uc.tasks, uc.catalog, affectedJobs, now); err != nil {
return err
}
@@ -159,6 +161,7 @@ func (uc *RenewLease) Execute(ctx context.Context, in RenewLeaseInput) (*domain.
type CompleteTask struct {
tasks TaskRepository
jobs JobRepository
catalog *workloads.Catalog
artifacts ArtifactRepository
workers WorkerRepository
results TaskResultRepository
@@ -170,12 +173,12 @@ type CompleteTask struct {
}
func NewCompleteTask(tasks TaskRepository, jobs JobRepository, artifacts ArtifactRepository,
workers WorkerRepository, results TaskResultRepository, tx TxManager, clock Clock, quorum int) *CompleteTask {
workers WorkerRepository, results TaskResultRepository, tx TxManager, clock Clock, quorum int, catalog *workloads.Catalog) *CompleteTask {
if quorum < 1 {
quorum = 2
}
return &CompleteTask{tasks: tasks, jobs: jobs, artifacts: artifacts, workers: workers,
results: results, tx: tx, clock: clock, quorum: quorum}
results: results, tx: tx, clock: clock, quorum: quorum, catalog: catalog}
}
// Execute applies the result and, when that was the job's last outstanding
@@ -229,7 +232,7 @@ func (uc *CompleteTask) Execute(ctx context.Context, in CompleteTaskInput) (*dom
if err := uc.tasks.Update(ctx, task); err != nil {
return err
}
return syncJobStatus(ctx, uc.jobs, uc.tasks, task.JobID, now)
return syncJobStatus(ctx, uc.jobs, uc.tasks, uc.catalog, task.JobID, now)
})
if err != nil {
return nil, err
@@ -268,7 +271,7 @@ func (uc *CompleteTask) recordVote(ctx context.Context, task *domain.Task, in Co
if err := uc.tasks.Update(ctx, task); err != nil {
return err
}
return syncJobStatus(ctx, uc.jobs, uc.tasks, task.JobID, now)
return syncJobStatus(ctx, uc.jobs, uc.tasks, uc.catalog, task.JobID, now)
}
// workerTrust reports whether the worker's results are accepted directly, and
@@ -324,10 +327,11 @@ type FailTask struct {
workers WorkerRepository
tx TxManager
clock Clock
catalog *workloads.Catalog
}
func NewFailTask(tasks TaskRepository, jobs JobRepository, workers WorkerRepository, tx TxManager, clock Clock) *FailTask {
return &FailTask{tasks: tasks, jobs: jobs, workers: workers, tx: tx, clock: clock}
func NewFailTask(tasks TaskRepository, jobs JobRepository, workers WorkerRepository, tx TxManager, clock Clock, catalog *workloads.Catalog) *FailTask {
return &FailTask{tasks: tasks, jobs: jobs, workers: workers, tx: tx, clock: clock, catalog: catalog}
}
// Execute delegates the requeue-or-terminate decision to Task.Fail, then keeps
@@ -351,7 +355,7 @@ func (uc *FailTask) Execute(ctx context.Context, in FailTaskInput) (*domain.Task
return err
}
out = task
return syncJobStatus(ctx, uc.jobs, uc.tasks, task.JobID, now)
return syncJobStatus(ctx, uc.jobs, uc.tasks, uc.catalog, task.JobID, now)
})
if err != nil {
return nil, err
@@ -362,14 +366,15 @@ func (uc *FailTask) Execute(ctx context.Context, in FailTaskInput) (*domain.Task
// --- ExpireLeases --------------------------------------------------------
type ExpireLeases struct {
tasks TaskRepository
jobs JobRepository
tx TxManager
clock Clock
tasks TaskRepository
jobs JobRepository
tx TxManager
clock Clock
catalog *workloads.Catalog
}
func NewExpireLeases(tasks TaskRepository, jobs JobRepository, tx TxManager, clock Clock) *ExpireLeases {
return &ExpireLeases{tasks: tasks, jobs: jobs, tx: tx, clock: clock}
func NewExpireLeases(tasks TaskRepository, jobs JobRepository, tx TxManager, clock Clock, catalog *workloads.Catalog) *ExpireLeases {
return &ExpireLeases{tasks: tasks, jobs: jobs, tx: tx, clock: clock, catalog: catalog}
}
// Execute reclaims elapsed tasks and persists the state of every affected job.
@@ -386,7 +391,7 @@ func (uc *ExpireLeases) Execute(ctx context.Context) (int64, error) {
if err != nil {
return err
}
return syncExpiredJobStatuses(ctx, uc.jobs, uc.tasks, affected, now)
return syncExpiredJobStatuses(ctx, uc.jobs, uc.tasks, uc.catalog, affected, now)
})
return int64(len(affected)), err
}
coordinator/internal/usecase/testcatalog_test.go
+62
View File
@@ -0,0 +1,62 @@
package usecase_test
import (
"testing"
"github.com/emil28092005/SciMesh/coordinator/internal/workloads"
)
// testCatalog loads the embedded workload catalog for usecase tests. The
// catalog is checked in and generated from the SDK library, so tests exercise
// the real validation contract.
func testCatalog() *workloads.Catalog {
catalog, err := workloads.Load()
if err != nil {
panic(err)
}
return catalog
}
func TestEmbeddedCatalogLoadsAndValidatesSearchParameters(t *testing.T) {
catalog := testCatalog()
if err := catalog.ValidateParameters("similarity-search", map[string]any{
"query_smiles": "CCO", "top_k": 10, "threshold_direction": "greater",
}); err != nil {
t.Fatalf("valid search parameters rejected: %v", err)
}
if err := catalog.ValidateParameters("molwt-filter", map[string]any{
"min_molwt": 100, "max_molwt": 600, "skip_invalid": true,
}); err != nil {
t.Fatalf("valid molwt parameters rejected: %v", err)
}
if err := catalog.ValidateParameters("nope", map[string]any{}); err == nil {
t.Error("unknown workload accepted")
}
if err := catalog.ValidateParameters("similarity-graph", map[string]any{"threshold": 0.7}); err != nil {
t.Errorf("graph parameters rejected: %v", err)
}
if !catalog.UploadReady("molwt-filter") {
t.Error("molwt-filter must be upload-ready")
}
if catalog.UploadReady("similarity-graph") {
t.Error("similarity-graph must not be upload-ready")
}
if got := catalog.Reduction("similarity-search"); got != "top-k" {
t.Errorf("search reduction = %q, want top-k", got)
}
if got := catalog.Reduction("descriptor-batch"); got != "ordered-concat" {
t.Errorf("descriptor reduction = %q, want ordered-concat", got)
}
for name, parameters := range map[string]map[string]any{
"both query fields": {"query_id": "CHEMBL1", "query_smiles": "CCO"},
"undeclared parameter": {"query_smiles": "CCO", "bogus": 1},
"bad top_k": {"query_smiles": "CCO", "top_k": -1},
"bad enum": {"query_smiles": "CCO", "threshold_direction": "sideways"},
"missing query": {},
"non-integer top_k": {"query_smiles": "CCO", "top_k": 1.5},
} {
if err := catalog.ValidateParameters("similarity-search", parameters); err == nil {
t.Errorf("%s accepted", name)
}
}
}
coordinator/internal/usecase/ui.go
+61 -17
View File
@@ -3,12 +3,15 @@ package usecase
import (
"context"
"fmt"
"sort"
"strings"
"time"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/authctx"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
"github.com/emil28092005/SciMesh/coordinator/internal/workloads"
)
// UIReadRepository is a read-only projection source for the local operator UI.
@@ -125,9 +128,14 @@ type JobDetailView struct {
Session *SessionView `json:"-"`
}
type Dashboard struct{ read UIReadRepository }
type Dashboard struct {
read UIReadRepository
catalog *workloads.Catalog
}
func NewDashboard(read UIReadRepository) *Dashboard { return &Dashboard{read: read} }
func NewDashboard(read UIReadRepository, catalog *workloads.Catalog) *Dashboard {
return &Dashboard{read: read, catalog: catalog}
}
func (d *Dashboard) Overview(ctx context.Context, limit int) (DashboardView, error) {
jobs, err := d.read.ListJobs(ctx, uiOwnerFilter(ctx), limit)
@@ -219,7 +227,7 @@ func (d *Dashboard) JobDetail(ctx context.Context, jobID uuid.UUID) (JobDetailVi
JobCard: jobCard(*job, tasks),
Tasks: make([]TaskCard, 0, len(tasks)),
Artifacts: make([]ArtifactCard, 0, len(artifacts)),
Parameters: uiParameters(job.Parameters),
Parameters: uiParameters(job.Parameters, d.catalog, job.Workload),
Session: sessionViewFrom(ctx),
}
for _, task := range tasks {
@@ -304,31 +312,67 @@ func jobCard(job domain.Job, tasks []domain.Task) JobCard {
return c
}
func uiParameters(parameters map[string]any) []ParameterCard {
keys := []struct {
key string
label string
}{
{"query_smiles", "Target SMILES"},
{"query_id", "Target ChEMBL ID"},
{"top_k", "Global top-k"},
{"threshold", "Similarity threshold"},
{"threshold_direction", "Threshold direction"},
func uiParameters(parameters map[string]any, catalog *workloads.Catalog, workload string) []ParameterCard {
labels := map[string]string{
"query_smiles": "Target SMILES",
"query_id": "Target ChEMBL ID",
"top_k": "Global top-k",
"threshold": "Similarity threshold",
"threshold_direction": "Threshold direction",
"min_molwt": "Minimum molecular weight",
"max_molwt": "Maximum molecular weight",
"skip_invalid": "Skip invalid molecules",
"block_size": "Block size",
}
keys := make([]string, 0, len(parameters))
declared := declaredParameterNames(catalog, workload)
for key := range parameters {
if declared != nil && !declared[key] {
// Only schema-declared scientific parameters may reach the browser;
// anything else could carry internal coordinator state.
continue
}
keys = append(keys, key)
}
sort.Strings(keys)
out := make([]ParameterCard, 0, len(keys))
for _, entry := range keys {
value, ok := parameters[entry.key]
for _, key := range keys {
value, ok := parameters[key]
if !ok {
continue
}
formatted, ok := formatUIParameter(value)
if ok {
out = append(out, ParameterCard{Label: entry.label, Value: formatted})
if !ok {
continue
}
label := labels[key]
if label == "" {
label = strings.ReplaceAll(key, "_", " ")
}
out = append(out, ParameterCard{Label: label, Value: formatted})
}
return out
}
func declaredParameterNames(catalog *workloads.Catalog, workload string) map[string]bool {
if catalog == nil || workload == "" {
return nil
}
item := catalog.ByName(workload)
if item == nil {
return nil
}
properties, ok := item.Parameters["properties"].(map[string]any)
if !ok {
return nil
}
declared := make(map[string]bool, len(properties))
for name := range properties {
declared[name] = true
}
return declared
}
func formatUIParameter(value any) (string, bool) {
switch typed := value.(type) {
case string:
coordinator/internal/usecase/ui_internal_test.go
+35 -2
View File
@@ -1,14 +1,22 @@
package usecase
import "testing"
import (
"testing"
"github.com/emil28092005/SciMesh/coordinator/internal/workloads"
)
func TestUIParametersAreAllowlisted(t *testing.T) {
catalog, err := workloads.Load()
if err != nil {
t.Fatal(err)
}
parameters := uiParameters(map[string]any{
"query_smiles": "CCO",
"top_k": float64(20),
"internal_storage_key": "must-not-reach-browser",
"nested": map[string]any{"secret": "no"},
})
}, catalog, "similarity-search")
if len(parameters) != 2 {
t.Fatalf("parameters = %#v, want only two allowlisted values", parameters)
}
@@ -17,3 +25,28 @@ func TestUIParametersAreAllowlisted(t *testing.T) {
t.Fatalf("parameters = %#v", parameters)
}
}
func TestUIParametersRenderEverySchemaDeclaredField(t *testing.T) {
catalog, err := workloads.Load()
if err != nil {
t.Fatal(err)
}
parameters := uiParameters(map[string]any{
"min_molwt": 100,
"max_molwt": 600,
"skip_invalid": true,
"secret_key": "no",
}, catalog, "molwt-filter")
if len(parameters) != 3 {
t.Fatalf("parameters = %#v, want three declared values", parameters)
}
labels := map[string]bool{}
for _, card := range parameters {
labels[card.Label] = true
}
for _, expected := range []string{"Minimum molecular weight", "Maximum molecular weight", "Skip invalid molecules"} {
if !labels[expected] {
t.Errorf("missing parameter card %q in %#v", expected, parameters)
}
}
}
coordinator/internal/usecase/ui_scope_test.go
+1 -1
View File
@@ -19,7 +19,7 @@ func newDashboard() (*usecase.Dashboard, *memstore.JobRepo) {
tasks := memstore.NewTaskRepo()
workers := memstore.NewWorkerRepo()
artifacts := memstore.NewArtifactRepo()
return usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, workers, artifacts)), jobs
return usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, workers, artifacts), testCatalog()), jobs
}
func ownedJob(t *testing.T, jobs *memstore.JobRepo, owner uuid.UUID) uuid.UUID {
coordinator/internal/usecase/ui_workers_scope_test.go
+1 -1
View File
@@ -17,7 +17,7 @@ func newDashboardWithWorkers() (*usecase.Dashboard, *memstore.WorkerRepo) {
tasks := memstore.NewTaskRepo()
workers := memstore.NewWorkerRepo()
artifacts := memstore.NewArtifactRepo()
return usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, workers, artifacts)), workers
return usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, workers, artifacts), testCatalog()), workers
}
func seedWorker(t *testing.T, workers *memstore.WorkerRepo, owner *uuid.UUID, name string) {
coordinator/internal/usecase/upload.go
+28 -62
View File
@@ -4,12 +4,12 @@ import (
"context"
"fmt"
"io"
"math"
"github.com/google/uuid"
"github.com/emil28092005/SciMesh/coordinator/internal/chunk"
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
"github.com/emil28092005/SciMesh/coordinator/internal/workloads"
)
// SubmitDataset accepts an uploaded dataset, splits it into shard artifacts, and
@@ -23,15 +23,16 @@ type SubmitDataset struct {
tx TxManager
clk Clock
maxAttempts int
catalog *workloads.Catalog
}
func NewSubmitDataset(blobs BlobStore, artifacts ArtifactRepository, jobs JobRepository,
tasks TaskRepository, tx TxManager, clk Clock, maxAttempts int) *SubmitDataset {
return &SubmitDataset{blobs: blobs, artifacts: artifacts, jobs: jobs, tasks: tasks, tx: tx, clk: clk, maxAttempts: maxAttempts}
tasks TaskRepository, tx TxManager, clk Clock, maxAttempts int, catalog *workloads.Catalog) *SubmitDataset {
return &SubmitDataset{blobs: blobs, artifacts: artifacts, jobs: jobs, tasks: tasks, tx: tx, clk: clk, maxAttempts: maxAttempts, catalog: catalog}
}
func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (SubmitDatasetResult, error) {
if err := validateUploadedWorkload(in.Workload, in.Parameters); err != nil {
if err := validateUploadedWorkload(uc.catalog, in.Workload, in.Parameters); err != nil {
return SubmitDatasetResult{}, err
}
if uc.maxAttempts < 1 {
@@ -86,7 +87,8 @@ func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (Su
putKeys = append(putKeys, art.StorageKey)
art.SetContent(ssum, ssize)
task, err := domain.NewShardTask(job.ID, index, in.Workload, art.ID, ssum, in.Parameters, uc.maxAttempts, now)
task, err := domain.NewShardTask(job.ID, index, in.Workload, art.ID, ssum,
taskParameterSubset(in.Parameters), uc.maxAttempts, now)
if err != nil {
return err
}
@@ -128,74 +130,38 @@ func (uc *SubmitDataset) Execute(ctx context.Context, in SubmitDatasetInput) (Su
}, nil
}
// validateUploadedWorkload is deliberately narrow until CTX-07/08/10 adds a
// typed distributed-workload registry. In particular, running similarity-graph
// independently per TSV shard is scientifically wrong: cross-shard pairs would
// be absent from the apparent graph.
func validateUploadedWorkload(workload string, parameters map[string]any) error {
if workload != "similarity-search" {
// validateUploadedWorkload checks the submitted workload against the embedded
// catalog: it must be an enabled, upload-ready workload whose parameters
// satisfy the declared JSON schema. Workloads that need planner-produced
// inputs (such as the graph block-pair shards) declare upload_ready=false and
// cannot be driven from a single uploaded dataset.
func validateUploadedWorkload(catalog *workloads.Catalog, workload string, parameters map[string]any) error {
if catalog == nil {
return domain.ErrInvalidInput
}
allowed := map[string]struct{}{
"query_smiles": {}, "top_k": {}, "threshold": {},
"threshold_direction": {}, "progress_every": {},
}
for key := range parameters {
if _, ok := allowed[key]; !ok {
return domain.ErrInvalidInput
}
}
query, ok := parameters["query_smiles"].(string)
if !ok || query == "" || len(query) > 200 {
if err := catalog.ValidateParameters(workload, parameters); err != nil {
return domain.ErrInvalidInput
}
if value, ok := parameters["top_k"]; ok && !isPositiveJSONInteger(value) {
return domain.ErrInvalidInput
}
if value, ok := parameters["progress_every"]; ok && !isNonNegativeJSONInteger(value) {
return domain.ErrInvalidInput
}
if value, ok := parameters["threshold"]; ok && !isUnitIntervalNumber(value) {
return domain.ErrInvalidInput
}
if value, ok := parameters["threshold_direction"]; ok && value != "greater" && value != "less" {
if !catalog.UploadReady(workload) {
return domain.ErrInvalidInput
}
return nil
}
func isPositiveJSONInteger(value any) bool { return isJSONInteger(value, false) }
func isNonNegativeJSONInteger(value any) bool { return isJSONInteger(value, true) }
func isJSONInteger(value any, allowZero bool) bool {
var n int64
switch v := value.(type) {
case int:
n = int64(v)
case int64:
n = v
case float64:
if math.Trunc(v) != v || v > math.MaxInt64 || v < math.MinInt64 {
return false
// taskParameterSubset drops coordinator-level keys from the parameters that
// are handed to workers. max_rows is a plan-time bound applied by the chunker
// here; a worker would reject it as outside its stage projection.
func taskParameterSubset(parameters map[string]any) map[string]any {
if _, present := parameters["max_rows"]; !present {
return parameters
}
subset := make(map[string]any, len(parameters)-1)
for key, value := range parameters {
if key != "max_rows" {
subset[key] = value
}
n = int64(v)
default:
return false
}
return n >= 0 && (allowZero || n > 0)
}
func isUnitIntervalNumber(value any) bool {
switch v := value.(type) {
case float64:
return !math.IsNaN(v) && !math.IsInf(v, 0) && v >= 0 && v <= 1
case int:
return v >= 0 && v <= 1
case int64:
return v >= 0 && v <= 1
default:
return false
}
return subset
}
// GetTaskInput resolves a task's input shard and opens it for streaming. The
coordinator/internal/usecase/usecase_test.go
+55 -7
View File
@@ -73,20 +73,20 @@ func newHarness() *harness {
}
tx := memstore.Tx{}
h.createJob = usecase.NewCreateJob(h.jobs, h.tasks, tx, h.clk)
h.submit = usecase.NewSubmitDataset(h.blobs, h.arts, h.jobs, h.tasks, tx, h.clk, 3)
h.claim = usecase.NewClaimTask(h.tasks, h.jobs, h.work, tx, h.clk, lease)
h.submit = usecase.NewSubmitDataset(h.blobs, h.arts, h.jobs, h.tasks, tx, h.clk, 3, testCatalog())
h.claim = usecase.NewClaimTask(h.tasks, h.jobs, h.work, tx, h.clk, lease, testCatalog())
h.renew = usecase.NewRenewLease(h.tasks, h.work, tx, h.clk, lease)
h.complete = usecase.NewCompleteTask(h.tasks, h.jobs, h.arts, h.work, h.taskResults, tx, h.clk, 2)
h.fail = usecase.NewFailTask(h.tasks, h.jobs, h.work, tx, h.clk)
h.complete = usecase.NewCompleteTask(h.tasks, h.jobs, h.arts, h.work, h.taskResults, tx, h.clk, 2, testCatalog())
h.fail = usecase.NewFailTask(h.tasks, h.jobs, h.work, tx, h.clk, testCatalog())
h.status = usecase.NewGetJobStatus(h.jobs, h.tasks)
h.results = usecase.NewListResults(h.tasks)
h.register = usecase.NewRegisterWorker(h.work, h.clk)
h.uploadArt = usecase.NewUploadArtifact(h.tasks, h.work, h.arts, h.blobs, tx, h.clk)
h.downloadArt = usecase.NewDownloadArtifact(h.arts, h.blobs)
h.getInput = usecase.NewGetTaskInput(h.tasks, h.arts, h.blobs)
h.expire = usecase.NewExpireLeases(h.tasks, h.jobs, tx, h.clk)
h.expire = usecase.NewExpireLeases(h.tasks, h.jobs, tx, h.clk, testCatalog())
h.cancel = usecase.NewCancelJob(h.jobs, h.tasks, tx, h.clk)
h.reduce = usecase.NewReduceJob(h.jobs, h.tasks, h.arts, h.blobs, tx, h.clk)
h.reduce = usecase.NewReduceJob(h.jobs, h.tasks, h.arts, h.blobs, tx, h.clk, testCatalog())
h.jobResult = usecase.NewGetJobResult(h.jobs, h.downloadArt)
return h
}
@@ -133,6 +133,44 @@ func TestSimilaritySearchReductionCreatesFinalArtifact(t *testing.T) {
}
}
func TestMolwtFilterReductionConcatenatesPartialsInOrder(t *testing.T) {
h := newHarness()
jobID := h.seedJob(t, "molwt-filter", 2)
if err := h.jobs.UpdateStatus(ctx, jobID, domain.JobRunning, nil); err != nil {
t.Fatal(err)
}
partials := []string{
"chembl_id,canonical_smiles\nA,CC\n",
"chembl_id,canonical_smiles\nB,CCCC\n",
}
for _, partial := range partials {
taskID, attempt := h.leaseOne(t, "w1", "molwt-filter")
art, err := h.uploadArt.Execute(ctx, usecase.UploadArtifactInput{TaskID: taskID, WorkerID: "w1", Attempt: attempt, Filename: "partial.csv", ContentType: "text/csv", Body: strings.NewReader(partial)})
if err != nil {
t.Fatal(err)
}
if _, err := h.complete.Execute(ctx, usecase.CompleteTaskInput{TaskID: taskID, WorkerID: "w1", Attempt: attempt, ResultArtifactID: art.ID}); err != nil {
t.Fatal(err)
}
}
if err := h.reduce.Execute(ctx, jobID); err != nil {
t.Fatal(err)
}
progress, err := h.status.Execute(ctx, jobID)
if err != nil || progress.Job.Status != domain.JobCompleted {
t.Fatalf("status=%s err=%v", progress.Job.Status, err)
}
art, body, err := h.jobResult.Execute(ctx, jobID)
if err != nil {
t.Fatal(err)
}
defer body.Close()
bytes, _ := io.ReadAll(body)
if art.Kind != domain.ArtifactFinalResult || string(bytes) != "chembl_id,canonical_smiles\nA,CC\nB,CCCC\n" {
t.Fatalf("unexpected final %q", bytes)
}
}
func TestSimilaritySearchReductionFailureIsSanitized(t *testing.T) {
h := newHarness()
jobID := h.seedJob(t, "similarity-search", 1)
@@ -817,8 +855,18 @@ func TestSubmitDatasetRejectsUnsupportedDistributedWorkloads(t *testing.T) {
Filename: "chembl.tsv", ContentType: "text/tab-separated-values",
Body: strings.NewReader("chembl_id\tcanonical_smiles\nA\tCC\n"),
})
if err != nil {
t.Errorf("query_id submission err = %v, want nil", err)
}
_, err = h.submit.Execute(ctx, usecase.SubmitDatasetInput{
Workload: "similarity-search", Parameters: map[string]any{
"query_id": "CHEMBL1", "query_smiles": "CCO",
}, RowsPerShard: 2,
Filename: "chembl.tsv", ContentType: "text/tab-separated-values",
Body: strings.NewReader("chembl_id\tcanonical_smiles\nA\tCC\n"),
})
if !errors.Is(err, domain.ErrInvalidInput) {
t.Errorf("query_id submission err = %v, want ErrInvalidInput", err)
t.Errorf("both query fields submission err = %v, want ErrInvalidInput", err)
}
}