From 7ad28b939ded6ba2e8775ecd521aec91df1312c2 Mon Sep 17 00:00:00 2001 From: Emil Date: Fri, 24 Jul 2026 17:28:16 +0300 Subject: [PATCH] Show live processing speed chart --- coordinator/internal/transport/http/server_test.go | 2 +- coordinator/internal/transport/http/templates/job.html | 10 ++++++++-- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/coordinator/internal/transport/http/server_test.go b/coordinator/internal/transport/http/server_test.go index 2fd9836..e43086e 100644 --- a/coordinator/internal/transport/http/server_test.go +++ b/coordinator/internal/transport/http/server_test.go @@ -472,7 +472,7 @@ func TestSimilaritySearchLifecyclePublishesFinalResult(t *testing.T) { } defer uiResponse.Body.Close() uiBody, _ := io.ReadAll(uiResponse.Body) - if uiResponse.StatusCode != http.StatusOK || !strings.Contains(string(uiBody), "Final result ready") || !strings.Contains(string(uiBody), "Preview CSV") { + if uiResponse.StatusCode != http.StatusOK || !strings.Contains(string(uiBody), "Final result ready") || !strings.Contains(string(uiBody), "Preview CSV") || !strings.Contains(string(uiBody), "Processing speed") { t.Fatalf("final UI = (%d, %q)", uiResponse.StatusCode, uiBody) } diff --git a/coordinator/internal/transport/http/templates/job.html b/coordinator/internal/transport/http/templates/job.html index 0234f40..f0091e6 100644 --- a/coordinator/internal/transport/http/templates/job.html +++ b/coordinator/internal/transport/http/templates/job.html @@ -7,6 +7,7 @@ SciMesh pipeline @@ -15,6 +16,8 @@

{{workloadLabel .Workload}}

Live pipeline

One job, shown from accepted input through its final coordinator-owned scientific result.

Live · refreshes every 2 seconds
{{statusLabel .Status}}

{{statusHint .Status}}

Completed shards are preserved.

{{.Completed}} of {{.Total}} shards complete

{{.Total}}total shards
{{.Completed}}completed
{{.Pending}}waiting
{{add .Leased .Running}}with workers
{{.Failed}}failed
{{.Cancelled}}stopped
+

Processing speed

Collecting live completion samples from this browser session.

Collecting…current pace
—peak observed
+

Pipeline stages

Each stage reflects coordinator state, not a simulated progress bar.

1TSV accepted

The coordinator stored the source and created shard tasks.

2Shards execute

{{.Completed}} of {{.Total}} candidate partitions are complete.

3Workers return CSVs

Workers upload a checked partial result for every completed shard.

4Global reduction

The coordinator waits until all shards are complete.

5Final CSV

Available only after deterministic reduction succeeds.

Run configuration

Allowlisted scientific parameters.

What is being computed?

{{range .Parameters}}
{{.Label}}{{.Value}}
{{else}}

No displayable parameters were supplied.

{{end}}

Result status

Safe operator guidance.

{{if .FinalResultAvailable}}

Final result ready

The coordinator merged shard candidates with exact scores and stored the global top-k CSV.

{{range .Artifacts}}{{if and (eq .Kind "final_result") .Downloadable}}Preview CSVDownload final CSV{{end}}{{end}}{{else if eq .Status "reducing"}}

Merging completed shards

The final candidate heap is being ranked now. This page will update when the CSV is stored.

{{else if eq .Status "failed"}}

Run needs attention

{{if .ErrorMessage}}{{.ErrorMessage}}{{else}}One or more shards could not produce a final result. Review the task table below.{{end}}

{{else}}

Waiting for the final result

Partial CSVs are diagnostics. They become one global result only after every shard completes and reduction succeeds.

{{end}}
@@ -33,9 +36,12 @@ 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 preview=text('a','Preview CSV','download');preview.href='/ui/jobs/'+encodeURIComponent(job.id)+'/artifacts/'+encodeURIComponent(final.id)+'/preview';const download=text('a','Download final CSV','download');download.href='/ui/jobs/'+encodeURIComponent(job.id)+'/artifacts/'+encodeURIComponent(final.id);card.append(preview,download)}}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 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 speedHistory=[{at:Date.now(),completed:Number({{.Completed}})}],speedWindow=15,speedLimit=90,svgNS='http://www.w3.org/2000/svg',speedSVG=(tag,attrs)=>{const node=document.createElementNS(svgNS,tag);for(const [key,value] of Object.entries(attrs))node.setAttribute(key,String(value));return node},rateLabel=rate=>rate.toFixed(1)+' shards/min'; + const renderSpeed=()=>{const chart=document.querySelector('#speed-chart'),rates=speedHistory.map((sample,index)=>{const base=speedHistory[Math.max(0,index-speedWindow)],minutes=Math.max((sample.at-base.at)/60000,1/60);return Math.max(0,(sample.completed-base.completed)/minutes)}),current=rates.at(-1),peak=Math.max(...rates),ready=speedHistory.length>1,width=600,height=160,left=42,right=16,top=16,bottom=28,plotWidth=width-left-right,plotHeight=height-top-bottom,maxRate=Math.max(1,peak);document.querySelector('#speed-current').textContent=ready?rateLabel(current):'Collecting…';document.querySelector('#speed-peak').textContent=ready?rateLabel(peak):'—';document.querySelector('#speed-caption').textContent=ready?'Completed shards per minute, calculated from live coordinator snapshots in this page visit.':'Collecting live completion samples from this browser session.';chart.replaceChildren();for(let row=0;row<=4;row++){const y=top+plotHeight*row/4,value=maxRate*(1-row/4);chart.append(speedSVG('line',{x1:left,y1:y,x2:width-right,y2:y,class:'speed-grid'}),speedSVG('text',{x:4,y:y+4,class:'speed-label'}));chart.lastChild.textContent=value.toFixed(1)}if(!ready)return;const points=rates.map((rate,index)=>{const x=left+plotWidth*index/Math.max(1,rates.length-1),y=top+plotHeight*(1-rate/maxRate);return{x,y}}),path=points.map((point,index)=>(index?'L':'M')+point.x.toFixed(1)+' '+point.y.toFixed(1)).join(' ');chart.append(speedSVG('path',{d:path,class:'speed-line'}));const latest=points.at(-1);chart.append(speedSVG('circle',{cx:latest.x,cy:latest.y,r:4,class:'speed-point'}))}; + const recordSpeed=job=>{const now=Date.now(),last=speedHistory.at(-1);if(last&&job.completed===last.completed&&now-last.at<1500)return;speedHistory.push({at:now,completed:job.completed});if(speedHistory.length>speedLimit)speedHistory.splice(0,speedHistory.length-speedLimit);renderSpeed()}; + 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);recordSpeed(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(); + renderSpeed();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();