ald-tao-gpu / server.py
TiGa-RCE's picture
Improve onboarding, useful defaults and continuation guidance
2a138ca verified
Raw History Blame
22.1 kB
"""Shared CPU routes and delivery API for hosted ALD training."""
import base64
import hashlib
import json
from pathlib import Path
import re
import threading
from fastapi import HTTPException, Request
from fastapi.responses import FileResponse, Response, HTMLResponse, JSONResponse
from starlette.concurrency import run_in_threadpool
from delivery import create_api, CapacityExceeded, InvalidAnalysisOutput, publish_worker_result
from worker import run_reported
from build_info import build_identity
from training_config import validate_config
from hf_identity import oauth_config
from builtin_tracks import TrackLibrary
from steam_leaderboard import compare_public_best_lap, milliseconds, SteamUnavailable
from bucket_storage import (BucketStorageError, MAX_REQUEST as BUCKET_MAX_REQUEST,
MAX_RECIPE_LIBRARY, run_bucket_operation, valid_bucket_id, valid_bundle_key,
valid_recipe_key, validate_recipe_library, AUTOSAVE_APP_KEY, winner_sidecar_filename)
from portable_bundle import (BundleError, load_bundle, MAX_BUNDLE, MAX_NETWORK,
MAX_ANALYSIS_ARCHIVE, MAX_ANALYSIS_FILE_BYTES)
ROOT=Path(__file__).resolve().parent
EVOLUTION_SIDECAR=re.compile(r'^[a-z0-9][a-z0-9-]{0,63}-(formula|rally|snowmobile|truck)-[0-9]{8}T[0-9]{6}Z-[0-9a-f]{8}(?:-(fitness-winner|fastest-lap-winner|clean-laps-winner))?\.evolution$')
def create_app(worker=run_reported,store=None,gpu_worker=None,track_library=None,require_analysis=False):
app=create_api(store)
admission=threading.BoundedSemaphore(1)
leaderboard_admission=threading.BoundedSemaphore(1)
bucket_admission=threading.BoundedSemaphore(1)
library=track_library if track_library is not None else TrackLibrary(ROOT/'builtin_tracks')
@app.middleware('http')
async def bucket_no_store(request:Request, call_next):
response=await call_next(request)
if request.url.path.startswith('/api/bucket/'):
response.headers['Cache-Control']='no-store'
return response
def bucket_token(request:Request):
authorization=request.headers.get('authorization','')
if not authorization.startswith('Bearer '):
raise HTTPException(401,'Storage credential required')
token=authorization[7:]
if not token or len(token)>4096 or any(ch.isspace() or ord(ch)<33 for ch in token):
raise HTTPException(401,'Storage credential required')
return token
async def bucket_body(request:Request, token:str, *, limit:int, fields:set[str]):
chunks=bytearray()
try:
async for chunk in request.stream():
if len(chunks)+len(chunk)>limit: raise HTTPException(413,'Storage request too large')
chunks.extend(chunk)
def unique_object(pairs):
result={}
for key,value in pairs:
if key in result: raise ValueError('duplicate key')
result[key]=value
return result
value=json.loads(chunks.decode('utf-8'),object_pairs_hook=unique_object)
if not isinstance(value,dict) or set(value)!=fields:
raise ValueError('invalid body')
except HTTPException:
raise
except Exception:
raise HTTPException(400,'Invalid storage request') from None
finally:
chunks.clear()
if not valid_bucket_id(value.get('bucket_id')):
raise HTTPException(400,'Invalid storage request')
return value
def bucket_http_error(exc:BucketStorageError):
status={'unauthorized':401,'forbidden':403,'not_found':404,'invalid_request':400,
'invalid_bundle':400,'storage_unavailable':502,'timeout':504}.get(exc.category,502)
detail=('Storage credential is invalid' if status==401 else
'Storage access is unavailable' if status==403 else
'Invalid portable bundle' if exc.category=='invalid_bundle' else
'Storage operation timed out' if status==504 else
'Storage operation failed')
return HTTPException(status,detail)
async def bucket_operation(request:Request, operation:str, *, fields:set[str], limit:int=4096):
token=bucket_token(request)
if not bucket_admission.acquire(blocking=False):
raise HTTPException(429,'Storage operation busy')
try:
body=await bucket_body(request,token,limit=limit,fields=fields)
bundle_raw=None
bundle_obj=None
bundle_files=None
if operation=='save':
bundle=body.get('bundle')
if not isinstance(bundle,str) or len(bundle.encode('utf-8'))>MAX_BUNDLE:
raise HTTPException(400,'Invalid portable bundle')
try: bundle_obj,bundle_files=load_bundle(bundle.encode('utf-8'))
except BundleError: raise HTTPException(400,'Invalid portable bundle') from None
bundle_raw=bundle.encode('utf-8')
call_args={'bundle_base64':base64.b64encode(bundle_raw).decode('ascii')}
elif operation=='recipes_save':
library=body.get('library')
if not isinstance(library,str): raise HTTPException(400,'Invalid recipe library')
try: library_raw=library.encode('utf-8')
except UnicodeEncodeError: raise HTTPException(400,'Invalid recipe library') from None
if len(library_raw)>MAX_RECIPE_LIBRARY:
raise HTTPException(413,'Recipe library too large')
try: validate_recipe_library(library_raw)
except Exception: raise HTTPException(400,'Invalid recipe library') from None
call_args={'library_base64':base64.b64encode(library_raw).decode('ascii')}
elif operation=='restore':
if not valid_bundle_key(body.get('key')):
raise HTTPException(400,'Invalid storage request')
call_args={'key':body['key']}
elif operation=='recipes_restore':
if not valid_recipe_key(body.get('key')):
raise HTTPException(400,'Invalid storage request')
call_args={'key':body['key']}
else:
call_args={}
try:
result=await run_in_threadpool(run_bucket_operation,operation,token,body['bucket_id'],**call_args)
except BucketStorageError as exc:
raise bucket_http_error(exc) from None
if operation=='save':
result['_validated_bundle_bytes']=bundle_raw
result['_validated_bundle_files']=bundle_files
result['_validated_bundle_has_receipt']='run_receipt' in bundle_obj['artifacts']
if operation=='recipes_save':
result['_validated_library_bytes']=library_raw
result['_validated_bucket_id']=body['bucket_id']
return result
finally:
bucket_admission.release()
@app.post('/api/bucket/connect')
async def bucket_connect(request:Request):
result=await bucket_operation(request,'connect',fields={'bucket_id'})
return JSONResponse(result,headers={'Cache-Control':'no-store'})
@app.post('/api/bucket/list')
async def bucket_list(request:Request):
result=await bucket_operation(request,'list',fields={'bucket_id'})
return JSONResponse(result,headers={'Cache-Control':'no-store'})
@app.post('/api/bucket/save')
async def bucket_save(request:Request):
result=await bucket_operation(request,'save',fields={'bucket_id','bundle'},limit=BUCKET_MAX_REQUEST)
raw=result.pop('_validated_bundle_bytes',None)
bundle_files=result.pop('_validated_bundle_files',None)
has_receipt=result.pop('_validated_bundle_has_receipt',False)
verified=result.get('verified_files')
required={'bundle.json'}
if has_receipt:
required|={'report.md','generation-history.csv'}
if isinstance(bundle_files,dict) and 'analysis_archive.gz' in bundle_files:
required.add('analysis-archive.gz')
required.update(name for name in bundle_files if name.startswith('analysis/'))
if 'analysis/stars.json' in bundle_files:
required.add('star-check.txt')
key_match=AUTOSAVE_APP_KEY.fullmatch(result.get('key',''))
if key_match is None:
raise HTTPException(502,'Storage operation failed')
for role in ('fitness','fastest_lap','clean_laps'):
inner=f'analysis/winners/{role}.evolution'
if inner in bundle_files:
required.add(winner_sidecar_filename(key_match.group('track'),key_match.group('vehicle'),
key_match.group('timestamp'),key_match.group('uuid')[:8],role))
evolution=[name for name in verified if isinstance(name,str)
and (match:=EVOLUTION_SIDECAR.fullmatch(name)) and match.group(2) is None] if isinstance(verified,dict) else []
if len(evolution)!=1 or not isinstance(verified,dict) or set(verified)!=(required|{evolution[0]}):
raise HTTPException(502,'Storage operation failed')
if not isinstance(raw,bytes) or not isinstance(bundle_files,dict):
raise HTTPException(502,'Storage operation failed')
network=bundle_files['selected_network.evolution']
safe_verified={}
total=0
per_file_limit=MAX_ANALYSIS_FILE_BYTES
for name,proof in verified.items():
if (not isinstance(proof,dict) or set(proof)!={'bytes','sha256'}
or type(proof['bytes']) is not int or not 0<=proof['bytes']<=per_file_limit
or not isinstance(proof['sha256'],str) or not re.fullmatch(r'[0-9a-f]{64}',proof['sha256'])):
raise HTTPException(502,'Storage operation failed')
total+=proof['bytes']
safe_verified[name]={'bytes':proof['bytes'],'sha256':proof['sha256']}
if (total>MAX_BUNDLE+MAX_NETWORK+320*1024*1024
or safe_verified['bundle.json']!={'bytes':len(raw),'sha256':hashlib.sha256(raw).hexdigest()}
or safe_verified[evolution[0]]!={'bytes':len(network),'sha256':hashlib.sha256(network).hexdigest()}):
raise HTTPException(502,'Storage operation failed')
analysis_archive=bundle_files.get('analysis_archive.gz')
if analysis_archive is not None and safe_verified.get('analysis-archive.gz')!={
'bytes':len(analysis_archive),'sha256':hashlib.sha256(analysis_archive).hexdigest()}:
raise HTTPException(502,'Storage operation failed')
for filename,content in bundle_files.items():
if filename.startswith('analysis/') and safe_verified.get(filename)!={
'bytes':len(content),'sha256':hashlib.sha256(content).hexdigest()}:
raise HTTPException(502,'Storage operation failed')
for role in ('fitness','fastest_lap','clean_laps'):
inner=f'analysis/winners/{role}.evolution'
key_match=AUTOSAVE_APP_KEY.fullmatch(result.get('key',''))
sidecar=(winner_sidecar_filename(key_match.group('track'),key_match.group('vehicle'),
key_match.group('timestamp'),key_match.group('uuid')[:8],role) if key_match else '')
if inner in bundle_files and safe_verified.get(sidecar)!={
'bytes':len(bundle_files[inner]),'sha256':hashlib.sha256(bundle_files[inner]).hexdigest()}:
raise HTTPException(502,'Storage operation failed')
return JSONResponse({key:result[key] for key in ('status','bucket_id','key','bytes','sha256')}|{'verified_files':safe_verified},headers={'Cache-Control':'no-store'})
@app.post('/api/bucket/restore')
async def bucket_restore(request:Request):
result=await bucket_operation(request,'restore',fields={'bucket_id','key'})
return Response(result['bundle_bytes'],media_type='application/json',headers={'Cache-Control':'no-store','X-Content-Type-Options':'nosniff'})
@app.post('/api/bucket/recipes/list')
async def bucket_recipes_list(request:Request):
result=await bucket_operation(request,'recipes_list',fields={'bucket_id'})
items=result.get('items')
if (not isinstance(items,list) or type(result.get('truncated')) is not bool
or type(result.get('scan_incomplete')) is not bool or len(items)>50):
raise HTTPException(502,'Storage operation failed')
clean=[]
for item in items:
if (not isinstance(item,dict) or set(item)!={'key','bytes','label','created_at'}
or not valid_recipe_key(item.get('key')) or type(item.get('bytes')) is not int
or not 0<=item['bytes']<=MAX_RECIPE_LIBRARY
or not isinstance(item.get('label'),str) or len(item['label'])>192
or (item.get('created_at') is not None and not isinstance(item.get('created_at'),str))):
raise HTTPException(502,'Storage operation failed')
clean.append(item)
return JSONResponse({'items':clean,'truncated':result['truncated'],
'scan_incomplete':result['scan_incomplete']},headers={'Cache-Control':'no-store'})
@app.post('/api/bucket/recipes/save')
async def bucket_recipes_save(request:Request):
result=await bucket_operation(request,'recipes_save',fields={'bucket_id','library'},
limit=MAX_RECIPE_LIBRARY*6+4096)
raw=result.pop('_validated_library_bytes',None)
expected_bucket=result.pop('_validated_bucket_id',None)
verified=result.pop('verified_bytes',None)
if (not isinstance(raw,bytes) or verified!=raw or result.get('status')!='saved_verified'
or result.get('bucket_id')!=expected_bucket or not valid_recipe_key(result.get('key'))
or result.get('bytes')!=len(raw) or result.get('sha256')!=hashlib.sha256(raw).hexdigest()):
raise HTTPException(502,'Storage operation failed')
return JSONResponse({key:result[key] for key in ('status','bucket_id','key','bytes','sha256')},
headers={'Cache-Control':'no-store'})
@app.post('/api/bucket/recipes/restore')
async def bucket_recipes_restore(request:Request):
result=await bucket_operation(request,'recipes_restore',fields={'bucket_id','key'})
raw=result.get('recipe_bytes')
if not isinstance(raw,bytes) or not valid_recipe_key(result.get('key')):
raise HTTPException(502,'Storage operation failed')
return Response(raw,media_type='application/json',headers={'Cache-Control':'no-store','X-Content-Type-Options':'nosniff'})
@app.get('/api/tracks')
def track_catalog():
return JSONResponse(library.catalog(),headers={'Cache-Control':'no-store'})
@app.post('/api/leaderboard/compare')
async def leaderboard_compare(request:Request):
data=bytearray()
async for chunk in request.stream():
if len(data)+len(chunk)>4096: raise HTTPException(413,'Request too large')
data.extend(chunk)
try:
body=json.loads(data)
if not isinstance(body,dict) or set(body)!={'track_id','track_sha256','vehicle_type','lap_seconds'}:
raise ValueError()
if not isinstance(body['track_id'],str) or not isinstance(body['track_sha256'],str): raise ValueError()
if type(body['vehicle_type']) is not int or not 0<=body['vehicle_type']<=3: raise ValueError()
milliseconds(body['lap_seconds'])
except Exception: raise HTTPException(400,'Invalid leaderboard comparison request') from None
finally: data.clear()
entry=next((track for track in library.catalog()['tracks'] if track['id']==body['track_id']),None)
if entry is None:
return {'status':'unsupported_track','message':'Steam comparison is available only for supported built-in tracks.'}
if body['track_sha256']!=entry['sha256']:
raise HTTPException(400,'Track identity did not match the built-in catalog')
if not leaderboard_admission.acquire(blocking=False):
return {'status':'busy','message':'Another Steam check is in progress. The completed result is unaffected.'}
try:
result=await run_in_threadpool(compare_public_best_lap,entry,body['lap_seconds'])
if result.get('status')=='ok':result['vehicle_type']=body['vehicle_type']
return result
except SteamUnavailable as error:
result={'status':'unavailable','failure_category':error.failure_category,
'message':'Steam could not provide a complete public leaderboard. The completed result is unaffected.'}
if error.failure_category == 'rate_limited':
result['message']='Steam is limiting leaderboard requests. Checks pause for one minute; your completed result is unaffected.'
if error.upstream_http_status is not None:
result['upstream_http_status']=error.upstream_http_status
return result
finally:
leaderboard_admission.release()
@app.get('/api/tracks/{identifier}/file')
def reference_track(identifier:str):
try:payload=library.read(identifier)
except KeyError:raise HTTPException(404,'Built-in track not found') from None
except (ValueError,OSError):raise HTTPException(503,'Built-in track validation failed') from None
return Response(payload,media_type='application/octet-stream',headers={'Cache-Control':'no-store'})
@app.get('/track-library.js')
def track_script():return FileResponse(ROOT/'track-library.js',media_type='text/javascript')
@app.get('/track-map.js')
def map_script():return FileResponse(ROOT/'track-map.js',media_type='text/javascript')
@app.get('/assets/track-atlas.json')
def track_atlas():return FileResponse(ROOT/'assets/track-atlas.json',media_type='application/json')
@app.get('/api/auth/config')
def auth_config():
return JSONResponse(oauth_config(),headers={'Cache-Control':'no-store'})
@app.get('/hf-auth.js')
def auth_script(): return FileResponse(ROOT/'hf-auth.js',media_type='text/javascript')
@app.get('/hf-oauth.js')
def oauth_script(): return FileResponse(ROOT/'vendor/hf-oauth/index.js',media_type='text/javascript')
@app.get('/api/capabilities')
def capabilities():
return {'cpu':True,'gpu':gpu_worker is not None,'default':'cpu','configurable_training':True,'build':build_identity()}
@app.get('/')
def index():
# Deliver the bootstrap with the authenticated document: private HF
# iframe subresource access can fail before an external script runs.
script=(ROOT/'client.js').read_text().replace('</script','<\\/script')
return HTMLResponse((ROOT/'index.html').read_text().replace(
'<script src="/client.js"></script>','<script>'+script+'</script>'),
headers={'Referrer-Policy':'no-referrer','Cache-Control':'no-store'})
@app.get('/client.js')
def script(): return FileResponse(ROOT/'client.js',media_type='text/javascript')
@app.get('/feedback')
def feedback(): return FileResponse(ROOT/'FEEDBACK.md',media_type='text/plain')
@app.get('/policy')
def policy(): return FileResponse(ROOT/'DATA_POLICY.md',media_type='text/plain')
@app.get('/help')
def help_page(): return FileResponse(ROOT/'help.html',media_type='text/html')
@app.post('/api/process')
async def process(request:Request):
if not admission.acquire(blocking=False): raise HTTPException(429,'Worker busy; try later')
try:
data=bytearray()
async for chunk in request.stream():
if len(data)+len(chunk)>3*1024*1024: raise HTTPException(413,'Request too large')
data.extend(chunk)
try:
body=json.loads(data)
automatic=body.get('automatic',True)
backend=body.get('backend','cpu')
if backend not in ('cpu','gpu'): raise ValueError()
if type(automatic) is not bool: raise ValueError()
track=base64.b64decode(body['track'],validate=True)
network=base64.b64decode(body['network'],validate=True)
if not track or not network or max(len(track),len(network))>1024*1024: raise ValueError()
except Exception: raise HTTPException(400,'Invalid input envelope') from None
finally: data.clear()
try:config=validate_config(body['settings']) if 'settings' in body else None
except ValueError as error:raise HTTPException(400,str(error)) from None
selected=worker if backend=='cpu' else gpu_worker
if selected is None: raise HTTPException(503,'GPU is unavailable; select CPU to run without a GPU lease')
try:
result=await run_in_threadpool(selected,track,network,config) if config is not None else await run_in_threadpool(selected,track,network)
payload,report=result if isinstance(result,tuple) else (result,None)
published,public_report=publish_worker_result(
app.state.results,payload,report,automatic,require_analysis=require_analysis)
return {**published,'backend':backend,'report':public_report}
except CapacityExceeded: raise HTTPException(503,'Result capacity reached') from None
except InvalidAnalysisOutput: raise HTTPException(422,'Processing failed; complete analysis could not be verified.') from None
except Exception: raise HTTPException(422,'Processing failed; no result published. Cleanup status requires verification.') from None
finally: admission.release()
return app
app=create_app()
if __name__=='__main__':
import uvicorn
uvicorn.run(app,host='127.0.0.1',port=47831,access_log=False)