Add serialized download queue with transfer speed and ETA

STUDIO.md

@@ -201,3 +201,9 @@ Die Update-Abfrage zeigt das aktuelle reguläre Release (`releases/latest`) zuer
 Unter GPU-Auswahl: Automatisch (maximaler Wert aus 512 MiB und 2,5 Prozent je GPU), Manuell (0–32768 MiB je ausgewählter GPU) oder Keine zusätzliche Reserve. Manuelle Werte werden per GPU-UUID gespeichert. Alte Profile verwenden weiter Automatisch. Die Reserve beeinflusst Fit-Ziel und Speicherentscheidung; Modell/KV/MTP sowie die konservative Projektor-Schätzung (Dateigröße plus 2 GiB Arbeitsbedarf) bleiben enthalten. Host-RAM-Grenzen bleiben bestehen. Ohne Reserve ist ein erfolgreicher Modellstart nicht garantiert.
 
 Für einen Versuch ohne CPU-Offloading zusätzlich „Vollständig auf GPU · sonst Start abbrechen“ wählen. Dann gibt es bei negativer Prognose keinen automatischen Layer-Rückfall. „Automatisch“ unter GPU-Ausführung erlaubt ihn weiterhin ausdrücklich. Der Testchat zeigt Layerlimit und Reserve je GPU. Gespeicherte Profile werden durch dieses Update nicht umgestellt.
+
+### Download-Warteschlange und Geschwindigkeit
+
+Während eines Downloads können bis zu 20 weitere Dateien eingereiht werden, auch kategorienübergreifend. Genau ein Transfer läuft; nach Abschluss, Fehler oder manuellem Abbruch startet der nächste. Doppelte Dateien werden abgewiesen. Unter Downloads lassen sich wartende Einträge entfernen. Die Speicherprüfung berücksichtigt noch ausstehende Bytes aller geplanten Transfers plus 10 GiB Reserve.
+
+Die API liefert `queued`, `bytes_per_second` und `eta_seconds`. Geschwindigkeit basiert auf einem gleitenden Zeitfenster; ETA ist eine Schätzung und enthält keine abschließende Dateiverarbeitung. Vor ausreichenden Messwerten bleibt die Anzeige offen. Beim Serverneustart werden laufende und wartende Einträge als unterbrochen markiert; keine automatische Wiederaufnahme.

catalog-ui.js

@@ -32,8 +32,8 @@ window.CatalogUI = (() => {
    el('hub-capacity').textContent = `Athena · ${size(state.free_bytes)} freier Speicher · 10 GiB bleiben als Reserve frei`;
    const jobs=downloads ? (state.downloads||[]).filter(j=>j.kind===kind) : state.job?.state==='downloading' ? [state.job] : [];
    const signature=JSON.stringify(jobs); if(signature===lastJobs)return; lastJobs=signature;
-   const labels={downloading:'Wird heruntergeladen',complete:'Download abgeschlossen',cancelled:'Download abgebrochen',failed:'Download fehlgeschlagen',interrupted:'Download unterbrochen'};
+   const labels={queued:'Warteschlange',downloading:'Wird heruntergeladen',complete:'Download abgeschlossen',cancelled:'Download abgebrochen',failed:'Download fehlgeschlagen',interrupted:'Download unterbrochen'};
-   el('hub-job').innerHTML=jobs.map(j=>{const percent=j.total>0?Math.min(100,Math.max(0,j.bytes/j.total*100)):0;return `<section class="card hub-operation"><div class="card-top"><span class="kicker">${escape(labels[j.state]||j.state)}</span><strong>${percent.toFixed(1)} %</strong></div><h3>${escape(j.repo)}</h3><p class="hub-filename">${escape(j.file)}</p><progress max="100" value="${percent}" aria-label="Download-Fortschritt"></progress><p>${size(j.bytes)} von ${size(j.total)}</p>${j.error?`<p class="warning">${escape(j.error)}</p>`:''}${j.state==='downloading'?'<button class="secondary" data-cancel>Download abbrechen</button>':`<button class="secondary" data-dismiss="${escape(j.id)}">Aus Liste entfernen</button><span class="small muted">Datei bleibt erhalten</span>`}</section>`;}).join('')||(downloads?'<div class="empty"><h3>Keine Downloads in dieser Liste</h3><p>Heruntergeladene Modelldateien findest du weiterhin in der Bibliothek.</p></div>':'');
+   el('hub-job').innerHTML=jobs.map(j=>{const percent=j.total>0?Math.min(100,Math.max(0,j.bytes/j.total*100)):0;return `<section class="card hub-operation"><div class="card-top"><span class="kicker">${escape(labels[j.state]||j.state)}</span><strong>${percent.toFixed(1)} %</strong></div><h3>${escape(j.repo)}</h3><p class="hub-filename">${escape(j.file)}</p><progress max="100" value="${percent}" aria-label="Download-Fortschritt"></progress><p>${size(j.bytes)} von ${size(j.total)}${j.state==='downloading'?` · ${Number.isFinite(j.bytes_per_second)?size(j.bytes_per_second)+'/s':'Geschwindigkeit wird ermittelt …'} · Restzeit: ${Number.isFinite(j.eta_seconds)?Math.ceil(j.eta_seconds/60)+' Min.':'–'}`:''}</p>${j.error?`<p class="warning">${escape(j.error)}</p>`:''}${j.state==='downloading'?'<button class="secondary" data-cancel>Download abbrechen</button>':`<button class="secondary" data-dismiss="${escape(j.id)}">${j.state==='queued'?'Aus Warteschlange entfernen':'Aus Liste entfernen'}</button><span class="small muted">${j.state==='queued'?'Download noch nicht gestartet':'Datei bleibt erhalten'}</span>`}</section>`;}).join('')||(downloads?'<div class="empty"><h3>Keine Downloads in dieser Liste</h3><p>Heruntergeladene Modelldateien findest du weiterhin in der Bibliothek.</p></div>':'');
    root.querySelectorAll('[data-cancel]').forEach(b=>b.onclick=async()=>{b.disabled=true;try{await api('/cancel',{});message('Abbruch angefordert; die Netzwerkabfrage kann bis zu 20 Sekunden dauern.');}catch(error){message(error.message);b.disabled=false;}});
    root.querySelectorAll('[data-dismiss]').forEach(b=>b.onclick=async()=>{b.disabled=true;try{await api('/dismiss',{id:b.dataset.dismiss});message('Eintrag ausgeblendet. Die Modelldatei bleibt in der Bibliothek.');await refresh();}catch(error){message(error.message);b.disabled=false;}});
   }
@@ -57,11 +57,12 @@ window.CatalogUI = (() => {
    if (!detail || !el('hub-download')) return;
    const chosen = el('hub-variants').querySelector('input:checked');
    const file = chosen ? detail.files[Number(chosen.value)] : null;
-   const busy = starting || state.job?.state === 'downloading';
+   const busy = starting;
+   const queued=file && (state.downloads||[]).some(j=>['queued','downloading'].includes(j.state)&&j.repo===detail.repo&&j.revision===detail.revision&&j.file===file.name);
    const exists = file && installed(file);
    const fitsDisk = file && Number.isFinite(state.free_bytes) && file.size <= state.free_bytes - 10*1024**3;
-   el('hub-download').disabled = !file || !!detail.gated || busy || exists || !fitsDisk;
+   el('hub-download').disabled = !file || !!detail.gated || busy || queued || exists || !fitsDisk;
-   el('hub-download').textContent = starting ? 'Download wird vorbereitet …' : exists ? 'Bereits in Bibliothek' : busy ? 'Ein Download läuft bereits' : 'Ausgewählte Datei herunterladen';
+   el('hub-download').textContent = starting ? 'Download wird vorbereitet …' : exists ? 'Bereits in Bibliothek' : queued ? 'Bereits im Downloadplan' : state.job?.state==='downloading' ? 'In Warteschlange einreihen' : 'Ausgewählte Datei herunterladen';
    el('hub-selection').textContent = file ? `${file.name} · ${size(file.size)}${!fitsDisk ? ' · Speicherplatz noch nicht bestätigt oder Reserve unterschritten' : ''}` : 'Wähle eine Datei aus der Liste.';
    if(el('hub-fit-result'))el('hub-fit-result').innerHTML=file?.assessment?`<strong>${escape(fitText(file.assessment))}</strong><p>${escape(file.assessment.explanation||'')}</p>${(file.assessment.gpus||[]).map(g=>`<p>${escape(g.name)}: ${size(g.capacity_bytes)} nutzbar insgesamt · ${size(g.free_bytes)} aktuell frei (nach Reserve)</p>`).join('')}${file.assessment.sampled_at?`<p>Messung: ${escape(new Date(file.assessment.sampled_at*1000).toLocaleTimeString('de-DE'))}</p>`:''}${file.assessment.deck_ram_limit_bytes?`<p>Deck-RAM-Limit: ${size(file.assessment.deck_ram_limit_bytes)}</p>`:''}`:'Wähle eine Datei für die Speicherprüfung. Gesamtbedarf der Laufzeit noch nicht berechnet.';
   }
@@ -83,7 +84,7 @@ window.CatalogUI = (() => {
      const chosen = el('hub-variants').querySelector('input:checked'); if (!chosen || starting) return;
      const request = {repo:detail.repo,revision:detail.revision,filename:detail.files[Number(chosen.value)].name,kind};
      starting = true; updateSelection();
-     try { await api('/download', request); message('Download gestartet. Den Fortschritt siehst du oberhalb der Galerie.'); await refresh(); }
+     try { const job=await api('/download', request); message(job.state==='queued'?'In Warteschlange eingereiht. Übersicht unter Downloads.':'Download gestartet. Fortschritt unter Downloads.'); await refresh(); }
      catch(error) { message(error.message); }
      finally { starting = false; if (root.isConnected) updateSelection(); }
     };

catalog.py

@@ -70,12 +70,12 @@ def capabilities(model):
 class Catalog:
     def __init__(self,root):
         self.root=Path(root);self.lock=threading.RLock();self.job=None;self.cancel=threading.Event()
-        self.cache={};self.cache_lock=threading.Lock();self.history=[]
+        self.cache={};self.cache_lock=threading.Lock();self.history=[];self.pending=[];self.closing=False;self.samples=[]
         path=self.root/'downloads.json'
         if path.exists():
             self.history=json.loads(path.read_text())
             for job in self.history:
-                if job['state']=='downloading':job.update(state='interrupted',error='Deck wurde neu gestartet. Datei erneut auswählen und herunterladen.')
+                if job['state'] in ('downloading','queued'):job.update(state='interrupted',error='Deck wurde neu gestartet. Datei erneut auswählen und herunterladen.')
         else:
             for entry in self.root.glob('*/entry.json'):
                 try:
@@ -91,6 +91,9 @@ class Catalog:
             job=next((x for x in self.history if x['id']==job_id),None)
             if not job:raise ValueError('Download nicht gefunden.')
             if job['state']=='downloading':raise ValueError('Laufenden Download zuerst abbrechen.')
+            if job['state']=='queued':
+                self.pending=[x for x in self.pending if x[0]['id']!=job_id]
+                job['state']='cancelled'
             job['dismissed']=True
             if self.job and self.job.get('id')==job_id:self.job=None
             self._save_history()
@@ -161,7 +164,8 @@ class Catalog:
     def start(self,repo,filename,revision,kind):
         if kind not in KINDS:raise ValueError('Ungültiger Bereich.')
         with self.lock:
-            if self.job and self.job['state']=='downloading':raise ValueError('Ein Download läuft bereits.')
+            if self.closing:raise ValueError('Deck wird beendet.')
+            if len(self.pending)>=20:raise ValueError('Warteschlange voll (20 Dateien).')
             with self.cache_lock:self.cache.pop(repo,None)
             data=self.files(repo)
             if data['gated']:raise ValueError('Zugangsbeschränkte Modelle werden noch nicht unterstützt.')
@@ -169,18 +173,25 @@ class Catalog:
             item=next((x for x in data['files'] if x['name']==filename),None)
             if not item:raise ValueError('Datei nicht verfügbar.')
             self.root.mkdir(parents=True,exist_ok=True,mode=0o700)
-            if item['size']>shutil.disk_usage(self.root).free-10*1024**3:raise ValueError('Nicht genug Platz mit 10 GiB freier Reserve.')
+            reserved=sum(x['total']-x['bytes'] for x in self.history if x['state'] in ('downloading','queued'))
+            if item['size']+reserved>shutil.disk_usage(self.root).free-10*1024**3:raise ValueError('Nicht genug Platz mit 10 GiB freier Reserve.')
             ident=hashlib.sha256((repo+revision+filename).encode()).hexdigest()
+            if any(x.get('entry_id')==ident and x['state'] in ('downloading','queued') for x in self.history):raise ValueError('Datei läuft bereits oder steht in der Warteschlange.')
             target=self.root/ident
             if (target/'entry.json').exists():raise ValueError('Datei bereits in der Bibliothek.')
             target.mkdir(exist_ok=True,mode=0o700)
-            self.cancel.clear()
+            job=dict(id=uuid.uuid4().hex,entry_id=ident,state='queued',repo=repo,file=filename,kind=kind,revision=revision,created_at=time.time(),bytes=0,total=item['size'],error=None,bytes_per_second=None,eta_seconds=None)
-            self.job=dict(id=uuid.uuid4().hex,entry_id=ident,state='downloading',repo=repo,file=filename,kind=kind,revision=revision,created_at=time.time(),bytes=0,total=item['size'],error=None)
+            self.history.append(job);self.pending.append((job,data,item,target,kind));self._next();self._save_history()
-            self.history.append(self.job);self._save_history()
+            return dict(job)
-            threading.Thread(target=self._download,args=(data,item,target,kind),daemon=True).start()
+    def _next(self):
-            return dict(self.job)
+        if self.closing or (self.job and self.job['state']=='downloading') or not self.pending:return
-    def stop(self):
+        self.job,data,item,target,kind=self.pending.pop(0)
-        self.cancel.set()
+        self.job['state']='downloading';self.cancel.clear();self.samples=[(time.monotonic(),0)]
+        threading.Thread(target=self._download,args=(data,item,target,kind),daemon=True).start()
+    def stop(self,shutdown=False):
+        with self.lock:
+            if shutdown:self.closing=True
+            self.cancel.set()
         return {'cancellation_requested':True}
     def _download(self,data,item,target,kind):
         partial=target/'download.part';dest=target/('model'+PurePosixPath(item['name']).suffix)
@@ -195,17 +206,22 @@ class Catalog:
                     received+=len(chunk)
                     if received>item['size'] or shutil.disk_usage(target).free<10*1024**3:raise ValueError('Größe oder Speicherreserve überschritten.')
                     out.write(chunk);digest.update(chunk)
-                    with self.lock:self.job['bytes']=received
+                    with self.lock:
+                        now=time.monotonic();self.samples.append((now,received))
+                        while len(self.samples)>2 and self.samples[1][0]<now-10:self.samples.pop(0)
+                        elapsed=now-self.samples[0][0];speed=(received-self.samples[0][1])/elapsed if elapsed>=.25 else None
+                        self.job.update(bytes=received,bytes_per_second=speed,eta_seconds=(item['size']-received)/speed if speed else None)
             if received!=item['size']:raise ValueError('Unvollständiger Download.')
             if item['sha256'] and digest.hexdigest()!=item['sha256']:raise ValueError('SHA-256-Prüfung fehlgeschlagen.')
             partial.replace(dest)
             entry=dict(repo=data['repo'],revision=data['revision'],file=item['name'],size=received,sha256=digest.hexdigest(),upstream_hash_verified=bool(item['sha256']),kind=kind,downloaded_at=time.time(),state='downloaded',runtime_ready=False)
             temp=target/'entry.tmp';temp.write_text(json.dumps(entry));temp.replace(target/'entry.json')
-            with self.lock:self.job['state']='complete'
+            terminal='complete';error=None
         except Exception as exc:
-            with self.lock:
+            terminal='cancelled' if isinstance(exc,InterruptedError) else 'failed'
-                self.job['state']='cancelled' if isinstance(exc,InterruptedError) else 'failed'
+            error=str(exc) if isinstance(exc,ValueError) else ('Abgebrochen.' if isinstance(exc,InterruptedError) else 'Download fehlgeschlagen; Verbindung oder Anbieter prüfen.')
-                self.job['error']=str(exc) if isinstance(exc,ValueError) else ('Abgebrochen.' if isinstance(exc,InterruptedError) else 'Download fehlgeschlagen; Verbindung oder Anbieter prüfen.')
             partial.unlink(missing_ok=True)
         finally:
-            with self.lock:self._save_history()
+            with self.lock:
+                self.job.update(state=terminal,error=error,bytes_per_second=None,eta_seconds=None)
+                self._save_history();self._next()

server.py

@@ -450,7 +450,7 @@ def main():
         server.image_runtime.stop()
         server.image_tests.stop()
         server.runtime.stop()
-        server.catalog.stop()
+        server.catalog.stop(shutdown=True)
         server.server_close()
 
 if __name__ == '__main__':

test_catalog.py

@@ -66,6 +66,31 @@ class CatalogTests(unittest.TestCase):
     with self.assertRaises(ValueError):c.start('a/b','m.gguf','a'*40,'chat')
     with patch('catalog.shutil.disk_usage',return_value=type('Disk',(),{'free':1024})()):
      with self.assertRaises(ValueError):c.start('a/b','m.gguf','b'*40,'chat')
+ def test_queue_progress_duplicate_and_removal(self):
+  import threading
+  release=threading.Event();progress=threading.Event()
+  class Stream(io.BytesIO):
+   def read(self,n=-1):
+    if self.tell()==0:
+     time.sleep(.3);return super().read(1)
+    progress.set();release.wait(3);return super().read(n)
+  with tempfile.TemporaryDirectory() as d:
+   c=Catalog(d);data=dict(repo='a/b',revision='a'*40,gated=False,files=[dict(name=n,size=2,sha256=None) for n in ['a.gguf','b.gguf','c.gguf']])
+   with patch.object(c,'files',return_value=data),patch('catalog.remote',side_effect=[Stream(b'aa'),io.BytesIO(b'bb')]),patch('catalog.shutil.disk_usage',return_value=type('Disk',(),{'free':100*1024**3})()):
+    first=c.start('a/b','a.gguf','a'*40,'chat');self.assertTrue(progress.wait(2))
+    status=c.status()['job'];self.assertGreater(status['bytes_per_second'],0);self.assertGreater(status['eta_seconds'],0)
+    second=c.start('a/b','b.gguf','a'*40,'chat');self.assertEqual(second['state'],'queued')
+    with self.assertRaises(ValueError):c.start('a/b','b.gguf','a'*40,'chat')
+    third=c.start('a/b','c.gguf','a'*40,'chat');c.dismiss(third['id'])
+    release.set()
+    for _ in range(200):
+     if len(c.status()['entries'])==2 and c.status()['job']['state']=='complete':break
+     time.sleep(.01)
+    self.assertEqual(len(c.status()['entries']),2);self.assertEqual(c.status()['job']['id'],second['id']);self.assertFalse(c.pending)
+ def test_shutdown_does_not_start_queued_file(self):
+  with tempfile.TemporaryDirectory() as d:
+   c=Catalog(d);c.pending=[('sentinel',)];c.stop(shutdown=True)
+   with patch('catalog.threading.Thread') as thread:c._next();thread.assert_not_called()
  def test_cancellation(self):
   with tempfile.TemporaryDirectory() as directory:
    c=Catalog(directory);c.job=dict(state='downloading');c.stop()