Merge pull request #359 from saphid/feat/framey-store-request-recovery

Recover Framey store jobs after lost responses

docs/framey-loader.md

@@ -203,12 +203,14 @@ the managed service is stopped; it does not restart the service.
 
 Install, Update, and Remove show the current job stage in the headset panel.
 The store verifies each downloaded archive and the headset applies it with a
-rollback if Decky cannot start. The panel remembers the current job ID for this
-browser session, so **Check result** can resume polling after a navigation or
-temporary connection error. After a polling error, Install, Update, and Remove
+rollback if Decky cannot start. The panel saves a random request ID before
+submitting a change. The server uses that ID as the job ID and returns the
+same job for a repeated, identical request while its result is retained;
+reusing the ID for a different change is rejected. **Check result** can resume
+polling even if the POST response is lost. After a connection error, Install, Update, and Remove
 stay disabled while the saved job's result is unknown; use **Check result**
 before starting another change.
-If Chromium denies session storage, the open panel keeps the accepted job ID in
+If Chromium denies session storage, the open panel keeps the request ID in
 memory and **Check result** still works after a connection error. A page reload
 loses that memory fallback; wait for the old job to finish and refresh installed
 status before starting another change.
@@ -216,9 +218,10 @@ Refresh the installed list before starting another
 operation after an error. The headset panel and desktop app send the hash of
 the version you saw; if the official store changes that version before the job
 starts, Frame Control stops before downloading or changing the headset. Refresh
-the store and review the new version before retrying. If Frame Control restarts, it cannot recover the old
-job result. The panel clears that job ID and asks you to refresh the installed
-list before making another change.
+the store and review the new version before retrying. Job results live in server
+memory for one hour after completion. If Frame Control restarts, or the request
+never reached it, the panel cannot recover the result. It clears that ID and
+asks you to refresh the installed list before making another change.
 
 An open plugin panel closes before its own Update or Remove starts, so its old
 frontend cannot call the changing backend. Open panel is unavailable while a

tests/e2e/test_framey.py

@@ -107,9 +107,16 @@ u=h/'.config/systemd/user/frame-control-decky.service'; u.parent.mkdir(parents=T
         self.assertEqual((catalogue[0]['version'], catalogue[0]['sha256']), ('1.0.0', HASHES[0]))
         status, _, _ = api('POST', '/api/decky/plugin', {'action':'install','id':999})
         self.assertEqual(status, 400)  # Third-party code needs explicit acknowledgement.
-        job = finished(ok('POST', '/api/decky/plugin',
-                          {'action':'install','id':999,'acknowledged':True}))
+        request = {'action':'install','id':999,'acknowledged':True,
+                   'requestId':'0123456789abcdef' * 2}
+        started = ok('POST', '/api/decky/plugin', request)
+        self.assertEqual(ok('POST', '/api/decky/plugin', request)['job'], started['job'])
+        status, conflict, _ = api('POST', '/api/decky/plugin', {**request, 'id':998})
+        self.assertEqual(status, 409)
+        self.assertIn('already used', conflict['error'])
+        job = finished(started)
         self.assertFalse(job.get('error'), job)
+        self.assertEqual(ok('POST', '/api/decky/plugin', request)['job'], started['job'])
         self.assertEqual(ok('GET', '/api/decky')['plugins'][0]['version'], '1.0.0')
         self.assertIn('Sample 1.0.0', ssh('cat ~/.local/share/frame-control/decky/plugins/sample-panel/dist/index.js'))
         receipt = HASHES[0]

tests/page/framey_store.mjs

@@ -1,4 +1,5 @@
 import assert from 'node:assert/strict';
+import {webcrypto} from 'node:crypto';
 import {readFileSync} from 'node:fs';
 import vm from 'node:vm';
 
@@ -17,6 +18,7 @@ const document = {getElementById(id) {if(!elements.has(id)) elements.set(id,new
   createElement() {return new Element();}, hidden:false};
 const session = new Map();
 const ctx = vm.createContext({document, location:{hash:'',pathname:'/'},
+  crypto:webcrypto,
   sessionStorage:{getItem:key=>session.get(key)||'',setItem:(key,value)=>session.set(key,value),removeItem:key=>session.delete(key)},
   history:{replaceState(){}}, confirm:()=>true,
   window:{addEventListener(){}}, setInterval(){}, setTimeout(){},
@@ -61,6 +63,45 @@ assert.equal(el('deckyJobProgress').value,100);
 assert.equal(el('deckyJobCheck').hidden,true);
 assert.equal(session.has('frameyPluginJob'),false);
 
+// If the server accepts the POST but its response is lost, the panel already
+// has the ID needed to find the job. It must not offer a second install.
+let lostJobId, lostPosts=0;
+ctx.mockApi=async (path,body) => {
+  if(path==='/api/decky/plugin'){
+    lostPosts++; lostJobId=body.requestId;
+    assert.match(lostJobId,/^[0-9a-f]{32}$/);
+    throw Error('POST response lost');
+  }
+  if(path.startsWith('/api/job')){
+    assert.equal(new URL(path,'http://localhost').searchParams.get('id'),lostJobId);
+    return {done:true,message:'Recovered install'};
+  }
+  if(path==='/api/decky')return {managed:true,running:true,plugins:[]};
+  throw Error(`unexpected ${path}`);
+};
+vm.runInContext('api=mockApi',ctx);
+await vm.runInContext('deckyChange("install",{id:2,name:"Plugin 2",sha256:"a".repeat(64)})',ctx);
+assert.equal(JSON.parse(session.get('frameyPluginJob')).job,lostJobId);
+assert.equal(el('deckyJobCheck').hidden,false);
+await vm.runInContext('deckyChange("install",{id:2,name:"Plugin 2",sha256:"a".repeat(64)})',ctx);
+assert.equal(lostPosts,1);
+await vm.runInContext('checkDeckyJob()',ctx);
+assert.equal(el('deckyJobStatus').textContent,'Recovered install');
+assert.equal(session.has('frameyPluginJob'),false);
+
+// A definite validation rejection never reserved a job on the server.
+ctx.mockApi=async path => {
+  if(path==='/api/decky/plugin'){
+    const error=Error('Invalid reviewed store version');error.status=400;throw error;
+  }
+  if(path==='/api/decky')return {managed:true,running:true,plugins:[]};
+  throw Error(`unexpected ${path}`);
+};
+vm.runInContext('api=mockApi',ctx);
+await vm.runInContext('deckyChange("install",{id:2,name:"Plugin 2",sha256:"bad"})',ctx);
+assert.equal(session.has('frameyPluginJob'),false);
+assert.equal(el('deckyJobCheck').hidden,true);
+
 // A stopped Decky service must still offer recovery by removing its owned plugin.
 ctx.mockApi=async path => path==='/api/decky' ?
   {managed:true,running:false,plugins:[{storeId:2,name:'Plugin 2',version:'1.0'}]} :
@@ -149,7 +190,7 @@ ctx.mockApi=async path => {
 };
 vm.runInContext('api=mockApi',ctx);
 await vm.runInContext('checkDeckyJob()',ctx);
-assert.match(el('deckyJobStatus').textContent,/app restarted and lost this job result/);
+assert.match(el('deckyJobStatus').textContent,/server has no record of this request/);
 assert.equal(el('deckyJobCheck').hidden,true);
 assert.equal(session.has('frameyPluginJob'),false);
 assert.equal(el('deckyStoreList').children[0].children[1].disabled,false);

tests/test_framey_store_jobs.py

@@ -0,0 +1,65 @@
+"""Store requests can reconnect when the headset panel loses the POST response."""
+import sandbox  # noqa: F401
+import sys
+import threading
+import time
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'ui'))
+import server
+
+
+class StoreJobs(unittest.TestCase):
+    def test_same_request_id_runs_once_and_conflicting_reuse_is_rejected(self):
+        request_id = 'a' * 32
+        started = threading.Event()
+        finish = threading.Event()
+        calls = []
+
+        def change(_ssh, action, plugin_id, report, reviewed_hash=None):
+            calls.append((action, plugin_id, reviewed_hash))
+            started.set()
+            self.assertTrue(finish.wait(5))
+            report('Installed', 100)
+            return {'message': 'Installed once'}
+
+        body = {'action': 'install', 'id': 2, 'sha256': 'b' * 64,
+                'acknowledged': True, 'requestId': request_id}
+        try:
+            with patch.object(server.frame_decky_store, 'change', side_effect=change):
+                first = server.decky_plugin(body)
+                self.assertTrue(started.wait(5))
+                self.assertEqual(server.decky_plugin(body), first)
+                for changed in ({'action': 'update'}, {'id': 3}, {'sha256': 'c' * 64}):
+                    with self.subTest(changed=changed), self.assertRaises(server.Failure) as error:
+                        server.decky_plugin({**body, **changed})
+                    self.assertEqual(error.exception.status, 409)
+                self.assertEqual(calls, [('install', 2, 'b' * 64)])
+                self.assertNotIn('_request', server.job_status('id=' + request_id))
+                finish.set()
+                for _ in range(100):
+                    if server.job_status('id=' + request_id)['done']:
+                        break
+                    time.sleep(0.01)
+                self.assertEqual(server.job_status('id=' + request_id)['message'], 'Installed once')
+                self.assertEqual(server.decky_plugin(body), first)
+                self.assertEqual(len(calls), 1)
+        finally:
+            finish.set()
+            with server._jobs_lock:
+                server._jobs.pop(request_id, None)
+
+    def test_invalid_request_id_does_not_start_work(self):
+        with patch.object(server.frame_decky_store, 'change') as change:
+            for request_id in (1, '', 'A' * 32, 'f' * 31, 'f' * 33, '../' * 11):
+                with self.subTest(request_id=request_id), self.assertRaises(server.Failure) as error:
+                    server.decky_plugin({'action': 'install', 'id': 2,
+                                         'acknowledged': True, 'requestId': request_id})
+                self.assertEqual(error.exception.status, 400)
+            change.assert_not_called()
+
+
+if __name__ == '__main__':
+    unittest.main()

ui/headset.html

@@ -72,7 +72,8 @@ const byId=id=>document.getElementById(id);
 let latestStatus=null, activeExtension=null, recordBusy=false;
 async function api(path,body){const response=await fetch(path,{method:body?'POST':'GET',
   headers:{'X-Frame-UI':key,'Content-Type':'application/json'},body:body?JSON.stringify(body):undefined});
-  const data=await response.json();if(!response.ok)throw Error(data.error||'Frame request failed');return data;}
+  const data=await response.json();if(!response.ok){const error=Error(data.error||'Frame request failed');
+    error.status=response.status;throw error;}return data;}
 async function refresh(){try{const status=latestStatus=await api('/api/status');
   byId('battery').textContent=status.battery?.percent==null?'Unavailable':`${status.battery.percent}%`;
   const free=status.disk?.home?.free;
@@ -107,9 +108,8 @@ async function refreshGames(){const list=byId('games');try{const data=await api(
     row.append(name,quit);list.append(row);}if(!data.running.length)list.textContent='No games are running.';
   }catch(error){list.textContent=error.message;}}
 function compatibilityText(plugin){const c=plugin.compatibility;return c ? `${c.status === 'na' ? 'N/A' : c.status}: ${c.reason}${c.verification?.length ? ' Verified: '+c.verification.join(', ')+'.' : ''}` : 'Unverified on Frame';}
-// Some Chromium profiles deny sessionStorage. Keep the accepted job ID in
-// memory so this open panel can still reconnect instead of offering a second
-// install while the first one may be running.
+// Some Chromium profiles deny sessionStorage. Keep the reserved request ID in
+// memory so this open panel can still reconnect if the POST response is lost.
 let deckyPendingJob=null;
 function savedDeckyJob(){
   if(deckyPendingJob)return deckyPendingJob;
@@ -124,6 +124,8 @@ function clearDeckyJob(){
   try{sessionStorage.removeItem('frameyPluginJob');}catch(_){}
 }
 function deckyJobPending(){return !!savedDeckyJob();}
+function newDeckyRequestId(){return Array.from(crypto.getRandomValues(new Uint8Array(16)),
+  byte=>byte.toString(16).padStart(2,'0')).join('');}
 let deckyState=null, deckyStore=[], deckyStoreIndex=0, deckyStoreRequest=0,
   deckyStatusRequest=0,
   activeFrameyId=null, activeFrameyReceipt=null, frameySession=null, deckyBusy=false;
@@ -200,7 +202,7 @@ async function followDeckyJob(job,action,name){
     catch(error){
       if(/no such job/i.test(error.message)){
         clearDeckyJob();
-        throw Error('The app restarted and lost this job result. Refresh installed plugins to see what changed.');
+        throw Error('The server has no record of this request. It may not have arrived, or the app may have restarted. Refresh installed plugins before trying again.');
       }
       throw error;
     }
@@ -224,14 +226,17 @@ async function deckyChange(action,plugin){if(deckyBusy)return;
   if(panelClosed)closeFrameyPanel();
   deckyJobView(`${action} ${plugin.name}`,
     panelClosed?'Starting… Old plugin panel closed until the change finishes.':'Starting…',0);
+  const requestId=newDeckyRequestId();
+  rememberDeckyJob({job:requestId,action,name:plugin.name});
   try{const started=await api('/api/decky/plugin',{action,id:plugin.id,
-    sha256:action==='remove'?undefined:plugin.sha256,acknowledged:true});
+    sha256:action==='remove'?undefined:plugin.sha256,acknowledged:true,requestId});
     if(!started.job)throw Error('The server did not return a job ID');
     rememberDeckyJob({job:started.job,action,name:plugin.name});
     await followDeckyJob(started.job,action,plugin.name);
     clearDeckyJob();
     if(panelNote)byId('deckyJobStatus').textContent+=panelNote;
-  }catch(error){deckyJobView(`${action} ${plugin.name}`,`Error: ${error.message}${panelNote}`,null,true);}
+  }catch(error){if(error.status>=400&&error.status<500)clearDeckyJob();
+    deckyJobView(`${action} ${plugin.name}`,`Error: ${error.message}${panelNote}`,null,true);}
   finally{deckyBusy=false;await refreshDecky();}
 }
 async function checkDeckyJob(){const saved=savedDeckyJob();

ui/server.py

@@ -345,11 +345,16 @@ def backfill_art(apps=(), titles=(), meant=None):
     return True
 
 
-def _new_job_locked(label, **fields):
+def _prune_jobs_locked():
     now = time.time()
     for j in [j for j, v in _jobs.items() if v["done"] and now - v["time"] > JOB_TTL]:
         del _jobs[j]
-    job = secrets.token_hex(8)
+
+
+def _new_job_locked(label, job_id=None, **fields):
+    _prune_jobs_locked()
+    now = time.time()
+    job = job_id or secrets.token_hex(8)
     _jobs[job] = {"label": label, "done": False, "error": None, "message": None, "result": None, "time": now, **fields}
     return job
 
@@ -376,10 +381,18 @@ def _run_job(job, label, work, progress):
             _jobs[job].update(fields, done=True, time=time.time())
 
 
-def start_job(label, work, progress=False):
+def start_job(label, work, progress=False, *, request_id=None, request=None):
     """Run work() in the background. It returns a dict with a "message"."""
     with _jobs_lock:
-        job = _new_job_locked(label)
+        if request_id is not None:
+            _prune_jobs_locked()
+            existing = _jobs.get(request_id)
+            if existing:
+                if existing.get('_request') != request:
+                    raise Failure('This plugin request ID was already used for a different change', 409)
+                return {"message": f"{label}…", "job": request_id}
+        job = _new_job_locked(label, job_id=request_id,
+                              **({'_request': request} if request_id is not None else {}))
     busy_thread(_run_job, job, label, work, progress).start()
     return {"message": f"{label}…", "job": job}
 
@@ -462,7 +475,7 @@ def queue_action(body):
 def job_status(query):
     with _jobs_lock:
         job = _jobs.get((parse_qs(query).get("id") or [""])[0])
-        snapshot = job and {k: v for k, v in job.items() if k != "time"}
+        snapshot = job and {k: v for k, v in job.items() if k not in ("time", "_request")}
     if not snapshot:
         raise Failure("no such job (the app may have restarted)", 404)
     return snapshot
@@ -572,9 +585,15 @@ def decky_plugin(body):
     if body.get('acknowledged') is not True:
         raise Failure('Confirm that store plugins run code as your Frame user', 400)
     reviewed_hash = body.get('sha256')
+    request_id = body.get('requestId')
+    if request_id is not None and (not isinstance(request_id, str) or
+                                   not re.fullmatch(r'[0-9a-f]{32}', request_id)):
+        raise Failure('Invalid plugin request ID', 400)
+    device = LINK.active_device()['id'] if LINK else FRAME
     return start_job('Decky plugin ' + action,
                      lambda report: frame_decky_store.change(ssh, action, plugin_id, report,
-                                                            reviewed_hash=reviewed_hash), progress=True)
+                                                            reviewed_hash=reviewed_hash), progress=True,
+                     request_id=request_id, request=(device, action, plugin_id, reviewed_hash))
 
 
 def framey_method_action(body):