Add safe artifact previews
This commit is contained in:
Emil
@@ -86,9 +86,10 @@ or exposes worker processes.
|
||||
For a hands-on run, open `/ui`, choose **New similarity search**, select a
|
||||
small ChEMBL-style TSV, then leave one or more `scimesh-worker` processes
|
||||
running in separate terminals. The detail page updates every two seconds and
|
||||
stops polling after a completed, failed, or cancelled job. Download the
|
||||
`final_result` artifact only after the job reaches **Completed**; shard partial
|
||||
CSVs remain available as diagnostics.
|
||||
stops polling after a completed, failed, or cancelled job. Use **Preview CSV**
|
||||
to inspect a bounded first page of a partial or completed final result before
|
||||
downloading it. The UI never exposes source datasets or shard inputs; partial
|
||||
CSVs remain available only as diagnostics.
|
||||
|
||||
`up` starts three services in order: Postgres waits until `pg_isready` passes, a
|
||||
one-shot `migrate` container applies the schema and exits, and only then does the
|
||||
|
||||
@@ -84,6 +84,7 @@ func run() error {
|
||||
GetJobResult: usecase.NewGetJobResult(jobRepo, usecase.NewDownloadArtifact(artifactRepo, blobStore)),
|
||||
GetTaskInput: usecase.NewGetTaskInput(taskRepo, artifactRepo, blobStore),
|
||||
Dashboard: usecase.NewDashboard(uiReadRepo),
|
||||
PreviewArtifact: usecase.NewPreviewArtifact(uiReadRepo, blobStore),
|
||||
}
|
||||
|
||||
// Background reapers are tracked so shutdown can wait for them. Without this
|
||||
|
||||
@@ -119,6 +119,10 @@ func (p JobProgress) DeriveStatus() JobStatus {
|
||||
switch {
|
||||
case p.Job.Status == JobCancelled:
|
||||
return JobCancelled
|
||||
case p.Job.Status == JobFailed:
|
||||
// A reducer may fail after every shard has completed. That terminal
|
||||
// failure must not be overwritten by an otherwise-complete task count.
|
||||
return JobFailed
|
||||
case p.Job.Status == JobReducing:
|
||||
return JobReducing
|
||||
case p.Total == 0:
|
||||
|
||||
@@ -82,6 +82,7 @@ func TestDeriveStatus(t *testing.T) {
|
||||
{"done and failed", JobProgress{Total: 3, Done: 2, Failed: 1}, JobFailed},
|
||||
{"failed but work remains", JobProgress{Total: 3, Pending: 1, Failed: 2}, JobRunning},
|
||||
{"cancelled job wins over task histogram", JobProgress{Job: Job{Status: JobCancelled}, Total: 3, Done: 1, Cancelled: 2}, JobCancelled},
|
||||
{"persisted reducer failure wins over completed tasks", JobProgress{Job: Job{Status: JobFailed}, Total: 3, Done: 3}, JobFailed},
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
|
||||
@@ -31,6 +31,7 @@ type UseCases struct {
|
||||
DownloadArtifact *usecase.DownloadArtifact
|
||||
GetTaskInput *usecase.GetTaskInput
|
||||
Dashboard *usecase.Dashboard
|
||||
PreviewArtifact *usecase.PreviewArtifact
|
||||
}
|
||||
|
||||
type Server struct {
|
||||
@@ -86,6 +87,7 @@ func (s *Server) Handler(token string, uiToken ...string) http.Handler {
|
||||
ui.HandleFunc("POST /ui/api/jobs/{job_id}/cancel", s.handleCancelJob)
|
||||
ui.HandleFunc("POST /ui/api/jobs/upload", s.handleUploadDataset)
|
||||
ui.HandleFunc("GET /ui/jobs/{job_id}/artifacts/{artifact_id}", s.handleUIArtifactDownload)
|
||||
ui.HandleFunc("GET /ui/jobs/{job_id}/artifacts/{artifact_id}/preview", s.handleUIArtifactPreview)
|
||||
mux.Handle("/ui", chain(ui, withRequestID, withAccessLog(s.log), withBasicAuth(uiToken[0]), withSameOrigin))
|
||||
mux.Handle("/ui/", chain(ui, withRequestID, withAccessLog(s.log), withBasicAuth(uiToken[0]), withSameOrigin))
|
||||
} else {
|
||||
|
||||
@@ -60,6 +60,7 @@ func newEnvWithUIToken(t *testing.T, ready func(context.Context) error, configur
|
||||
DownloadArtifact: downloadArtifact,
|
||||
GetTaskInput: usecase.NewGetTaskInput(tasks, arts, blobs),
|
||||
Dashboard: usecase.NewDashboard(memstore.NewUIReadRepo(jobs, tasks, work, arts)),
|
||||
PreviewArtifact: usecase.NewPreviewArtifact(memstore.NewUIReadRepo(jobs, tasks, work, arts), blobs),
|
||||
}
|
||||
worker, err := uc.RegisterWorker.Execute(context.Background(), usecase.RegisterWorkerInput{
|
||||
Name: "test-worker", Capabilities: []string{"w", "similarity-search"},
|
||||
@@ -489,6 +490,50 @@ func TestSimilaritySearchLifecyclePublishesFinalResult(t *testing.T) {
|
||||
if jsonResponse.StatusCode != http.StatusOK || detail["final_result_available"] != true {
|
||||
t.Fatalf("final UI JSON = (%d, %v)", jsonResponse.StatusCode, detail)
|
||||
}
|
||||
artifacts := detail["artifacts"].([]any)
|
||||
var finalID string
|
||||
for _, raw := range artifacts {
|
||||
artifact := raw.(map[string]any)
|
||||
if artifact["kind"] == "final_result" && artifact["downloadable"] == true {
|
||||
finalID = artifact["id"].(string)
|
||||
break
|
||||
}
|
||||
}
|
||||
if finalID == "" {
|
||||
t.Fatalf("artifacts = %v, want downloadable final result", artifacts)
|
||||
}
|
||||
var inputID string
|
||||
for _, raw := range artifacts {
|
||||
artifact := raw.(map[string]any)
|
||||
if artifact["kind"] == "input" {
|
||||
inputID = artifact["id"].(string)
|
||||
break
|
||||
}
|
||||
}
|
||||
if inputID == "" {
|
||||
t.Fatalf("artifacts = %v, want input artifact", artifacts)
|
||||
}
|
||||
inputRequest, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/jobs/"+jobID+"/artifacts/"+inputID, nil)
|
||||
inputRequest.SetBasicAuth("operator", uiToken)
|
||||
inputResponse, err := http.DefaultClient.Do(inputRequest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer inputResponse.Body.Close()
|
||||
if inputResponse.StatusCode != http.StatusNotFound {
|
||||
t.Fatalf("UI input download = %d, want 404", inputResponse.StatusCode)
|
||||
}
|
||||
previewRequest, _ := http.NewRequestWithContext(context.Background(), "GET", e.ts.URL+"/ui/jobs/"+jobID+"/artifacts/"+finalID+"/preview", nil)
|
||||
previewRequest.SetBasicAuth("operator", uiToken)
|
||||
previewResponse, err := http.DefaultClient.Do(previewRequest)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer previewResponse.Body.Close()
|
||||
previewBody, _ := io.ReadAll(previewResponse.Body)
|
||||
if previewResponse.StatusCode != http.StatusOK || !strings.Contains(string(previewBody), "Final result preview") || !strings.Contains(string(previewBody), "0.900000") {
|
||||
t.Fatalf("final preview = (%d, %q)", previewResponse.StatusCode, previewBody)
|
||||
}
|
||||
}
|
||||
|
||||
func TestForeignArtifactResultConflict(t *testing.T) {
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
{{define "artifact-preview.html"}}
|
||||
<!doctype html>
|
||||
<html lang="en">
|
||||
<head>
|
||||
<meta charset="utf-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||
<title>SciMesh · artifact preview</title>
|
||||
<style>
|
||||
:root{color:#e4eeff;background:#08111f;font:16px/1.5 Inter,ui-sans-serif,system-ui,sans-serif;color-scheme:dark}body{margin:0;background:radial-gradient(circle at 10% -5%,#173f76 0,transparent 34rem),#08111f}.page{max-width:1180px;margin:auto;padding:32px 20px 56px}a{color:#8ab5ff}.back{text-decoration:none}h1{margin:18px 0 4px;color:#f4f8ff;font-size:1.6rem;word-break:break-word}.muted{color:#9cb0cb}.notice{margin:16px 0;padding:15px 17px;border:1px solid #aa8844;border-radius:10px;background:#302610;color:#f2dd9a}.table-wrap{overflow-x:auto;border:1px solid #294662;border-radius:10px;background:#0d1a2cdc;margin-top:16px}table{width:100%;border-collapse:collapse}td,th{padding:10px 12px;border-bottom:1px solid #203a55;text-align:left;vertical-align:top;white-space:pre-wrap;word-break:break-word}th{color:#9cb9dc;font-size:.78rem;text-transform:uppercase;letter-spacing:.06em;background:#10253d}tr:last-child td{border:0}.empty{padding:24px;text-align:center;color:#9cb0cb}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<main class="page">
|
||||
<a class="back" href="/ui/jobs/{{.JobID}}">← Back to job</a>
|
||||
<h1>Preview: {{.Filename}}</h1>
|
||||
{{if .Diagnostic}}
|
||||
<p class="muted">Diagnostic preview — a shard-level partial result, not the final molecular-search answer. At most {{.RowLimit}} rows and {{.ByteLimit}} bytes are read from storage.</p>
|
||||
{{else}}
|
||||
<p class="muted">Final result preview. At most {{.RowLimit}} rows and {{.ByteLimit}} bytes are read from storage.</p>
|
||||
{{end}}
|
||||
{{if not .Previewable}}
|
||||
<div class="notice">{{.Reason}}</div>
|
||||
{{else}}
|
||||
{{if .Truncated}}<div class="notice">Truncated: showing at most the first {{.RowLimit}} rows or {{.ByteLimit}} bytes. Download the artifact for its full contents.</div>{{end}}
|
||||
<div class="table-wrap">
|
||||
<table>
|
||||
<tr>{{range .Headers}}<th>{{.}}</th>{{end}}</tr>
|
||||
{{range .Rows}}<tr>{{range .}}<td>{{.}}</td>{{end}}</tr>{{else}}<tr><td class="empty" colspan="99">No data rows.</td></tr>{{end}}
|
||||
</table>
|
||||
</div>
|
||||
{{end}}
|
||||
</main>
|
||||
</body>
|
||||
</html>
|
||||
{{end}}
|
||||
@@ -21,7 +21,7 @@
|
||||
|
||||
<section><div class="section-head"><h2>Shard activity</h2><p id="task-caption">Every task is one input partition. The table refreshes while work is in progress.</p></div><div class="table-wrap"><table><thead><tr><th>Shard</th><th>State</th><th>Attempt</th><th>Worker / lease</th><th>Outcome</th></tr></thead><tbody id="tasks">{{range .Tasks}}<tr><td>#{{.ChunkIndex}}</td><td><span class="badge badge-{{statusClass .Status}}">{{statusLabel .Status}}</span></td><td>{{.Attempt}} / {{.MaxAttempts}}</td><td>{{if .LeaseOwner}}<strong>{{.LeaseOwner}}</strong>{{if .LeaseExpiresAt}}<br><small class="muted">lease until {{time .LeaseExpiresAt}}</small>{{end}}{{else}}<span class="muted">—</span>{{end}}</td><td class="error">{{if .ErrorCode}}<strong>{{taskErrorLabel .ErrorCode}}</strong><br><small>{{taskErrorHint .ErrorCode}}</small>{{else if eq .Status "completed"}}<span class="muted">Partial CSV uploaded</span>{{else}}<span class="muted">—</span>{{end}}</td></tr>{{else}}<tr><td colspan="5" class="empty">No shard tasks are present yet.</td></tr>{{end}}</tbody></table></div></section>
|
||||
|
||||
<section><div class="section-head"><h2>Coordinator artifacts</h2><p>All files stay coordinator-owned; checksums make downloads auditable.</p></div><div id="artifacts" class="artifact-grid">{{range .Artifacts}}<article class="artifact {{if eq .Kind "final_result"}}artifact-final{{end}}"><div class="artifact-type">{{if .Diagnostic}}Partial result · diagnostic{{else}}{{.Kind}}{{end}}</div><strong>{{.Filename}}</strong><span class="muted">{{bytes .SizeBytes}}</span><code>SHA-256 {{.SHA256}}</code>{{if .Downloadable}}<a href="/ui/jobs/{{$.ID}}/artifacts/{{.ID}}">Download CSV</a>{{end}}</article>{{else}}<div class="empty">Artifacts appear here as the coordinator stores input, shards, partial results, and the final CSV.</div>{{end}}</div></section>
|
||||
<section><div class="section-head"><h2>Coordinator artifacts</h2><p>All files stay coordinator-owned; checksums make downloads auditable.</p></div><div id="artifacts" class="artifact-grid">{{range .Artifacts}}<article class="artifact {{if eq .Kind "final_result"}}artifact-final{{end}}"><div class="artifact-type">{{if .Diagnostic}}Partial result · diagnostic{{else}}{{.Kind}}{{end}}</div><strong>{{.Filename}}</strong><span class="muted">{{bytes .SizeBytes}}</span><code>SHA-256 {{.SHA256}}</code>{{if .Downloadable}}<a href="/ui/jobs/{{$.ID}}/artifacts/{{.ID}}/preview">Preview CSV</a><a href="/ui/jobs/{{$.ID}}/artifacts/{{.ID}}">Download CSV</a>{{end}}</article>{{else}}<div class="empty">Artifacts appear here as the coordinator stores input, shards, partial results, and the final CSV.</div>{{end}}</div></section>
|
||||
<details class="technical"><summary>Technical details</summary><p>Job ID: <code id="job-id">{{.ID}}</code><button class="copy" id="copy-job-id" type="button">Copy</button><br>Created: {{time .CreatedAt}}<br>Workload: <code>{{.Workload}}</code></p></details>
|
||||
</main>
|
||||
<script>
|
||||
@@ -32,7 +32,7 @@
|
||||
const renderParameters=parameters=>{const holder=document.querySelector('#parameters');holder.replaceChildren();if(!parameters.length){holder.append(text('p','No displayable parameters were supplied.'));return}for(const parameter of parameters){const row=text('div',undefined,'parameter');row.append(text('span',parameter.label),text('code',parameter.value));holder.append(row)}};
|
||||
const renderResult=job=>{const card=document.querySelector('#result-card');card.className='run-note';card.replaceChildren();if(job.final_result_available){card.classList.add('result');card.append(text('h3','Final result ready'),text('p','The coordinator merged shard candidates with exact scores and stored the global top-k CSV.'));const final=(job.artifacts||[]).find(a=>a.kind==='final_result'&&a.downloadable);if(final){const link=text('a','Download final CSV','download');link.href='/ui/jobs/'+encodeURIComponent(job.id)+'/artifacts/'+encodeURIComponent(final.id);card.append(link)}}else if(job.status==='reducing'){card.append(text('h3','Merging completed shards'),text('p','The final candidate heap is being ranked now. This page will update when the CSV is stored.'))}else if(job.status==='failed'){card.classList.add('alert');card.append(text('h3','Run needs attention'),text('p',job.error_message||'One or more shards could not produce a final result. Review the task table below.'))}else{card.append(text('h3','Waiting for the final result'),text('p','Partial CSVs are diagnostics. They become one global result only after every shard completes and reduction succeeds.'))}};
|
||||
const renderTasks=tasks=>{const holder=document.querySelector('#tasks');holder.replaceChildren();if(!tasks.length){const row=document.createElement('tr'),cell=text('td','No shard tasks are present yet.','empty');cell.colSpan=5;row.append(cell);holder.append(row);return}for(const task of tasks){const row=document.createElement('tr'),info=statusInfo[task.status]||[task.status,'waiting',''];row.append(text('td','#'+task.chunk_index));const state=text('td'),badge=text('span',info[0],'badge badge-'+info[1]);state.append(badge);row.append(state,text('td',task.attempt+' / '+task.max_attempts));const worker=text('td');if(task.lease_owner){worker.append(text('strong',task.lease_owner));if(task.lease_expires_at){worker.append(document.createElement('br'),text('small','lease until '+fmtTime(task.lease_expires_at),'muted'))}}else worker.append(text('span','—','muted'));row.append(worker);const outcome=text('td',undefined,'error');if(task.error_code){const explanation=taskError[task.error_code]||['Task needs attention','Check the worker terminal for the original error.'];outcome.append(text('strong',explanation[0]),document.createElement('br'),text('small',explanation[1]))}else if(task.status==='completed')outcome.append(text('span','Partial CSV uploaded','muted'));else outcome.append(text('span','—','muted'));row.append(outcome);holder.append(row)}};
|
||||
const renderArtifacts=job=>{const holder=document.querySelector('#artifacts');holder.replaceChildren();const artifacts=job.artifacts||[];if(!artifacts.length){holder.append(text('div','Artifacts appear here as the coordinator stores input, shards, partial results, and the final CSV.','empty'));return}for(const artifact of artifacts){const card=text('article',undefined,'artifact'+(artifact.kind==='final_result'?' artifact-final':''));card.append(text('div',artifact.diagnostic?'Partial result · diagnostic':artifact.kind,'artifact-type'),text('strong',artifact.filename),text('span',fmtBytes(artifact.size_bytes),'muted'),text('code','SHA-256 '+artifact.sha256));if(artifact.downloadable){const link=text('a',artifact.kind==='final_result'?'Download final CSV':'Download CSV');link.href='/ui/jobs/'+encodeURIComponent(job.id)+'/artifacts/'+encodeURIComponent(artifact.id);card.append(link)}holder.append(card)}};
|
||||
const renderArtifacts=job=>{const holder=document.querySelector('#artifacts');holder.replaceChildren();const artifacts=job.artifacts||[];if(!artifacts.length){holder.append(text('div','Artifacts appear here as the coordinator stores input, shards, partial results, and the final CSV.','empty'));return}for(const artifact of artifacts){const card=text('article',undefined,'artifact'+(artifact.kind==='final_result'?' artifact-final':''));card.append(text('div',artifact.diagnostic?'Partial result · diagnostic':artifact.kind,'artifact-type'),text('strong',artifact.filename),text('span',fmtBytes(artifact.size_bytes),'muted'),text('code','SHA-256 '+artifact.sha256));if(artifact.downloadable){const preview=text('a','Preview CSV');preview.href='/ui/jobs/'+encodeURIComponent(job.id)+'/artifacts/'+encodeURIComponent(artifact.id)+'/preview';const download=text('a',artifact.kind==='final_result'?'Download final CSV':'Download CSV');download.href='/ui/jobs/'+encodeURIComponent(job.id)+'/artifacts/'+encodeURIComponent(artifact.id);card.append(preview,download)}holder.append(card)}};
|
||||
const render=job=>{const info=statusInfo[job.status]||[job.status,'waiting','Status reported by the coordinator.'],active=(job.leased||0)+(job.running||0),done=(job.completed||0)+(job.failed||0)+(job.cancelled||0),badge=document.querySelector('#status');badge.textContent=info[0];badge.className='badge badge-'+info[1];document.querySelector('#hint').textContent=info[2];document.querySelector('#progress-bar').style.width=pct(job)+'%';document.querySelector('#progress').textContent=job.completed+' of '+job.total+' shards complete'+(job.failed?' · '+job.failed+' failed':'')+(job.cancelled?' · '+job.cancelled+' stopped':'');for(const [key,value] of Object.entries({total:job.total,completed:job.completed,pending:job.pending,active,failed:job.failed,cancelled:job.cancelled}))document.querySelector('#'+key).textContent=value;document.querySelector('#stop-wrap').classList.toggle('hidden',!(job.status==='pending'||job.status==='running'));document.querySelector('#task-caption').textContent=done+' of '+job.total+' task states are terminal. The table refreshes while work is in progress.';renderStages(job);renderParameters(job.parameters||[]);renderResult(job);renderTasks(job.tasks||[]);renderArtifacts(job)};
|
||||
const stop=document.querySelector('#stop-job');if(stop)stop.addEventListener('click',async()=>{if(!confirm('Stop this job? Unfinished shards will be cancelled.'))return;stop.disabled=true;try{const response=await fetch('/ui/api/jobs/'+id+'/cancel',{method:'POST'});if(!response.ok)throw Error();await refresh()}catch(_){stop.disabled=false;alert('Unable to stop this job.')}});document.querySelector('#copy-job-id').addEventListener('click',async()=>{try{await navigator.clipboard.writeText(id);document.querySelector('#copy-job-id').textContent='Copied'}catch(_){}});
|
||||
let timer;const refresh=async()=>{try{const response=await fetch('/ui/api/jobs/'+id,{headers:{Accept:'application/json'}});if(!response.ok)throw Error();const job=await response.json();render(job);document.querySelector('#refresh-state').textContent=terminal.has(job.status)?'Final coordinator state':'Live · updated just now';if(terminal.has(job.status)&&timer){clearInterval(timer);timer=undefined}}catch(_){document.querySelector('#refresh-state').textContent='Connection interrupted · trying again automatically'}};const start=()=>{if(!timer&&!document.hidden)timer=setInterval(refresh,2000)};document.addEventListener('visibilitychange',()=>{if(document.hidden&&timer){clearInterval(timer);timer=undefined}else start()});start();
|
||||
|
||||
@@ -285,7 +285,7 @@ func (s *Server) handleUIArtifactDownload(w http.ResponseWriter, r *http.Request
|
||||
}
|
||||
ctx, cancel := s.reqCtx(r)
|
||||
defer cancel()
|
||||
belongs, err := s.uc.Dashboard.ArtifactBelongsToJob(ctx, jobID, artifactID)
|
||||
belongs, err := s.uc.Dashboard.DownloadableArtifactBelongsToJob(ctx, jobID, artifactID)
|
||||
if err != nil {
|
||||
s.writeError(w, r, err)
|
||||
return
|
||||
@@ -311,3 +311,30 @@ func (s *Server) handleUIArtifactDownload(w http.ResponseWriter, r *http.Request
|
||||
w.Header().Set("X-Checksum-SHA256", art.SHA256)
|
||||
_, _ = io.Copy(w, body)
|
||||
}
|
||||
|
||||
// handleUIArtifactPreview renders a bounded CSV preview. Its use case owns
|
||||
// the job-scoped access rule, including the requirement that a final artifact
|
||||
// is the persisted result of a completed job.
|
||||
func (s *Server) handleUIArtifactPreview(w http.ResponseWriter, r *http.Request) {
|
||||
jobID, ok := s.uiJobID(w, r)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
artifactID, err := uuid.Parse(r.PathValue("artifact_id"))
|
||||
if err != nil {
|
||||
s.writeError(w, r, domain.ErrInvalidInput)
|
||||
return
|
||||
}
|
||||
if s.uc.PreviewArtifact == nil {
|
||||
http.NotFound(w, r)
|
||||
return
|
||||
}
|
||||
ctx, cancel := s.reqCtx(r)
|
||||
defer cancel()
|
||||
view, err := s.uc.PreviewArtifact.Execute(ctx, jobID, artifactID)
|
||||
if err != nil {
|
||||
s.writeError(w, r, err)
|
||||
return
|
||||
}
|
||||
s.renderUI(w, "artifact-preview.html", view)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,146 @@
|
||||
package usecase
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/csv"
|
||||
"errors"
|
||||
"io"
|
||||
"mime"
|
||||
"strings"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
|
||||
)
|
||||
|
||||
// The preview is deliberately a diagnostic aid, never a full artifact
|
||||
// viewer. These limits bound both memory use and storage reads.
|
||||
const (
|
||||
previewMaxRows = 30
|
||||
previewMaxBytes = 64 * 1024
|
||||
)
|
||||
|
||||
// ArtifactPreviewView is the safe, bounded data rendered by the operator UI.
|
||||
// It deliberately contains neither storage keys nor worker-local details.
|
||||
type ArtifactPreviewView struct {
|
||||
JobID string
|
||||
ArtifactID string
|
||||
Filename string
|
||||
Diagnostic bool
|
||||
Previewable bool
|
||||
Reason string
|
||||
Headers []string
|
||||
Rows [][]string
|
||||
Truncated bool
|
||||
RowLimit int
|
||||
ByteLimit int64
|
||||
}
|
||||
|
||||
// PreviewArtifact reads the beginning of a job-scoped CSV result. Partial
|
||||
// results are diagnostic; a final result is available only after the reducer
|
||||
// has persisted it as this job's completed result.
|
||||
type PreviewArtifact struct {
|
||||
read UIReadRepository
|
||||
blobs BlobStore
|
||||
}
|
||||
|
||||
func NewPreviewArtifact(read UIReadRepository, blobs BlobStore) *PreviewArtifact {
|
||||
return &PreviewArtifact{read: read, blobs: blobs}
|
||||
}
|
||||
|
||||
func (p *PreviewArtifact) Execute(ctx context.Context, jobID, artifactID uuid.UUID) (ArtifactPreviewView, error) {
|
||||
job, err := p.read.GetJob(ctx, jobID)
|
||||
if err != nil {
|
||||
return ArtifactPreviewView{}, err
|
||||
}
|
||||
artifacts, err := p.read.ListArtifactsByJob(ctx, jobID)
|
||||
if err != nil {
|
||||
return ArtifactPreviewView{}, err
|
||||
}
|
||||
|
||||
var artifact *domain.Artifact
|
||||
for i := range artifacts {
|
||||
if artifacts[i].ID == artifactID {
|
||||
artifact = &artifacts[i]
|
||||
break
|
||||
}
|
||||
}
|
||||
if artifact == nil || !previewableArtifact(*job, *artifact) {
|
||||
// Use one response for an unknown artifact, another job's artifact, and
|
||||
// an artifact that is not yet public. This avoids leaking its state.
|
||||
return ArtifactPreviewView{}, domain.ErrArtifactNotFound
|
||||
}
|
||||
|
||||
view := ArtifactPreviewView{
|
||||
JobID: jobID.String(),
|
||||
ArtifactID: artifact.ID.String(),
|
||||
Filename: artifact.Filename,
|
||||
Diagnostic: artifact.Kind == domain.ArtifactPartialResult,
|
||||
RowLimit: previewMaxRows,
|
||||
ByteLimit: previewMaxBytes,
|
||||
}
|
||||
if !isCSVArtifact(artifact) {
|
||||
view.Reason = "This artifact is not a CSV file, so it cannot be shown as text here. Download it instead."
|
||||
return view, nil
|
||||
}
|
||||
if artifact.SizeBytes == 0 {
|
||||
view.Reason = "This artifact is empty."
|
||||
return view, nil
|
||||
}
|
||||
|
||||
body, err := p.blobs.Open(ctx, artifact.StorageKey)
|
||||
if err != nil {
|
||||
return ArtifactPreviewView{}, err
|
||||
}
|
||||
defer func() { _ = body.Close() }()
|
||||
|
||||
limited := &io.LimitedReader{R: body, N: previewMaxBytes}
|
||||
reader := csv.NewReader(limited)
|
||||
reader.FieldsPerRecord = -1 // a byte limit may end inside a record
|
||||
|
||||
headers, err := reader.Read()
|
||||
if err != nil {
|
||||
view.Reason = "This artifact could not be read as CSV."
|
||||
return view, nil
|
||||
}
|
||||
view.Headers = append([]string(nil), headers...)
|
||||
view.Rows = make([][]string, 0, previewMaxRows)
|
||||
for len(view.Rows) < previewMaxRows {
|
||||
record, readErr := reader.Read()
|
||||
if readErr != nil {
|
||||
if !errors.Is(readErr, io.EOF) {
|
||||
view.Truncated = true
|
||||
}
|
||||
break
|
||||
}
|
||||
view.Rows = append(view.Rows, append([]string(nil), record...))
|
||||
}
|
||||
|
||||
if artifact.SizeBytes > previewMaxBytes {
|
||||
view.Truncated = true
|
||||
} else if len(view.Rows) == previewMaxRows {
|
||||
if _, readErr := reader.Read(); readErr == nil {
|
||||
view.Truncated = true
|
||||
}
|
||||
}
|
||||
view.Previewable = true
|
||||
return view, nil
|
||||
}
|
||||
|
||||
func previewableArtifact(job domain.Job, artifact domain.Artifact) bool {
|
||||
if artifact.Kind == domain.ArtifactPartialResult {
|
||||
return true
|
||||
}
|
||||
return artifact.Kind == domain.ArtifactFinalResult &&
|
||||
job.Status == domain.JobCompleted &&
|
||||
job.ResultArtifactID != nil &&
|
||||
*job.ResultArtifactID == artifact.ID
|
||||
}
|
||||
|
||||
func isCSVArtifact(artifact *domain.Artifact) bool {
|
||||
mediaType, _, err := mime.ParseMediaType(artifact.ContentType)
|
||||
if err == nil && strings.EqualFold(mediaType, "text/csv") {
|
||||
return true
|
||||
}
|
||||
return strings.HasSuffix(strings.ToLower(artifact.Filename), ".csv")
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
package usecase_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
|
||||
"github.com/emil28092005/SciMesh/coordinator/internal/domain"
|
||||
"github.com/emil28092005/SciMesh/coordinator/internal/memstore"
|
||||
"github.com/emil28092005/SciMesh/coordinator/internal/usecase"
|
||||
)
|
||||
|
||||
func newPreviewHarness() (*usecase.PreviewArtifact, *memstore.JobRepo, *memstore.ArtifactRepo, *memstore.BlobStore) {
|
||||
jobs := memstore.NewJobRepo()
|
||||
tasks := memstore.NewTaskRepo()
|
||||
workers := memstore.NewWorkerRepo()
|
||||
artifacts := memstore.NewArtifactRepo()
|
||||
blobs := memstore.NewBlobStore()
|
||||
return usecase.NewPreviewArtifact(memstore.NewUIReadRepo(jobs, tasks, workers, artifacts), blobs), jobs, artifacts, blobs
|
||||
}
|
||||
|
||||
func previewJob(t *testing.T, jobs *memstore.JobRepo, status domain.JobStatus) uuid.UUID {
|
||||
t.Helper()
|
||||
job := &domain.Job{ID: uuid.New(), Workload: "similarity-search", Status: status, CreatedAt: time.Now().UTC()}
|
||||
if err := jobs.Insert(context.Background(), job); err != nil {
|
||||
t.Fatalf("insert preview job: %v", err)
|
||||
}
|
||||
return job.ID
|
||||
}
|
||||
|
||||
func previewArtifact(t *testing.T, artifacts *memstore.ArtifactRepo, blobs *memstore.BlobStore, jobID uuid.UUID, kind domain.ArtifactKind, filename, contentType, contents string) uuid.UUID {
|
||||
t.Helper()
|
||||
id := uuid.New()
|
||||
sha, size, err := blobs.Put(context.Background(), id.String(), strings.NewReader(contents))
|
||||
if err != nil {
|
||||
t.Fatalf("store preview artifact: %v", err)
|
||||
}
|
||||
artifact := &domain.Artifact{ID: id, JobID: jobID, Kind: kind, Filename: filename, StorageKey: id.String(), ContentType: contentType, SizeBytes: size, SHA256: sha, CreatedAt: time.Now().UTC()}
|
||||
if err := artifacts.Insert(context.Background(), artifact); err != nil {
|
||||
t.Fatalf("insert preview artifact: %v", err)
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func TestPreviewArtifactRendersBoundedCSV(t *testing.T) {
|
||||
preview, jobs, artifacts, blobs := newPreviewHarness()
|
||||
jobID := previewJob(t, jobs, domain.JobRunning)
|
||||
var csv strings.Builder
|
||||
csv.WriteString("chembl_id,score\n")
|
||||
for i := 0; i < 40; i++ {
|
||||
csv.WriteString("CHEMBL" + strconv.Itoa(i) + ",0.9\n")
|
||||
}
|
||||
artifactID := previewArtifact(t, artifacts, blobs, jobID, domain.ArtifactPartialResult, "partial.csv", "text/csv; charset=utf-8", csv.String())
|
||||
|
||||
view, err := preview.Execute(context.Background(), jobID, artifactID)
|
||||
if err != nil {
|
||||
t.Fatalf("preview: %v", err)
|
||||
}
|
||||
if !view.Previewable || !view.Diagnostic || !view.Truncated || len(view.Rows) != 30 || view.Headers[0] != "chembl_id" || view.Rows[0][0] != "CHEMBL0" {
|
||||
t.Fatalf("unexpected preview: %+v", view)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPreviewArtifactCapsStorageReadAndHandlesInvalidCSV(t *testing.T) {
|
||||
preview, jobs, artifacts, blobs := newPreviewHarness()
|
||||
jobID := previewJob(t, jobs, domain.JobRunning)
|
||||
// Fewer than 30 oversized records force the byte cap, rather than the row
|
||||
// cap, to stop parsing.
|
||||
large := "id,value\n" + strings.Repeat("row,"+strings.Repeat("x", 5*1024)+"\n", 20)
|
||||
largeID := previewArtifact(t, artifacts, blobs, jobID, domain.ArtifactPartialResult, "large.csv", "text/csv", large)
|
||||
view, err := preview.Execute(context.Background(), jobID, largeID)
|
||||
if err != nil || !view.Previewable || !view.Truncated || len(view.Rows) > 30 {
|
||||
t.Fatalf("large preview = (%+v, %v)", view, err)
|
||||
}
|
||||
invalidID := previewArtifact(t, artifacts, blobs, jobID, domain.ArtifactPartialResult, "broken.csv", "text/csv", "\"unterminated")
|
||||
invalid, err := preview.Execute(context.Background(), jobID, invalidID)
|
||||
if err != nil || invalid.Previewable || invalid.Reason == "" {
|
||||
t.Fatalf("invalid preview = (%+v, %v)", invalid, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPreviewArtifactRejectsOtherJobsAndNonResults(t *testing.T) {
|
||||
preview, jobs, artifacts, blobs := newPreviewHarness()
|
||||
jobA := previewJob(t, jobs, domain.JobRunning)
|
||||
jobB := previewJob(t, jobs, domain.JobRunning)
|
||||
partialID := previewArtifact(t, artifacts, blobs, jobA, domain.ArtifactPartialResult, "partial.csv", "text/csv", "a,b\n1,2\n")
|
||||
if _, err := preview.Execute(context.Background(), jobB, partialID); !errors.Is(err, domain.ErrArtifactNotFound) {
|
||||
t.Fatalf("cross-job preview error = %v", err)
|
||||
}
|
||||
inputID := previewArtifact(t, artifacts, blobs, jobA, domain.ArtifactInput, "input.csv", "text/csv", "a,b\n1,2\n")
|
||||
if _, err := preview.Execute(context.Background(), jobA, inputID); !errors.Is(err, domain.ErrArtifactNotFound) {
|
||||
t.Fatalf("input preview error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPreviewArtifactExposesOnlyPersistedCompletedFinalResult(t *testing.T) {
|
||||
preview, jobs, artifacts, blobs := newPreviewHarness()
|
||||
jobID := previewJob(t, jobs, domain.JobReducing)
|
||||
finalID := previewArtifact(t, artifacts, blobs, jobID, domain.ArtifactFinalResult, "final.csv", "text/csv", "rank,chembl_id\n1,CHEMBL1\n")
|
||||
if _, err := preview.Execute(context.Background(), jobID, finalID); !errors.Is(err, domain.ErrArtifactNotFound) {
|
||||
t.Fatalf("uncompleted final preview error = %v", err)
|
||||
}
|
||||
if err := jobs.CompleteWithResult(context.Background(), jobID, finalID, time.Now().UTC()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
view, err := preview.Execute(context.Background(), jobID, finalID)
|
||||
if err != nil || !view.Previewable || view.Diagnostic {
|
||||
t.Fatalf("completed final preview = (%+v, %v)", view, err)
|
||||
}
|
||||
if err := jobs.FailReduction(context.Background(), jobID, "reducer_failed", "final result reduction failed", time.Now().UTC()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := preview.Execute(context.Background(), jobID, finalID); !errors.Is(err, domain.ErrArtifactNotFound) {
|
||||
t.Fatalf("failed reducer preview error = %v", err)
|
||||
}
|
||||
}
|
||||
@@ -180,7 +180,7 @@ func (d *Dashboard) JobDetail(ctx context.Context, jobID uuid.UUID) (JobDetailVi
|
||||
}
|
||||
for _, artifact := range artifacts {
|
||||
diagnostic := artifact.Kind == domain.ArtifactPartialResult
|
||||
downloadable := diagnostic || (artifact.Kind == domain.ArtifactFinalResult && out.Status == string(domain.JobCompleted))
|
||||
downloadable := previewableArtifact(*job, artifact)
|
||||
out.Artifacts = append(out.Artifacts, ArtifactCard{ID: artifact.ID.String(), Kind: string(artifact.Kind), Filename: artifact.Filename, SizeBytes: artifact.SizeBytes, SHA256: artifact.SHA256, Downloadable: downloadable, Diagnostic: diagnostic})
|
||||
if artifact.Kind == domain.ArtifactFinalResult && downloadable {
|
||||
out.FinalResultAvailable = true
|
||||
@@ -189,13 +189,21 @@ func (d *Dashboard) JobDetail(ctx context.Context, jobID uuid.UUID) (JobDetailVi
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (d *Dashboard) ArtifactBelongsToJob(ctx context.Context, jobID, artifactID uuid.UUID) (bool, error) {
|
||||
// DownloadableArtifactBelongsToJob applies the same policy used by the UI
|
||||
// projection: partial diagnostics and the persisted final result are public to
|
||||
// the operator; source inputs and shards are not exposed through a guessed UI
|
||||
// URL.
|
||||
func (d *Dashboard) DownloadableArtifactBelongsToJob(ctx context.Context, jobID, artifactID uuid.UUID) (bool, error) {
|
||||
job, err := d.read.GetJob(ctx, jobID)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
artifacts, err := d.read.ListArtifactsByJob(ctx, jobID)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
for _, a := range artifacts {
|
||||
if a.ID == artifactID {
|
||||
if a.ID == artifactID && previewableArtifact(*job, a) {
|
||||
return true, nil
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user