diff --git a/make_post_sell/lib/karaoke.py b/make_post_sell/lib/karaoke.py index 9cafc4e..c4507d7 100644 --- a/make_post_sell/lib/karaoke.py +++ b/make_post_sell/lib/karaoke.py @@ -295,6 +295,198 @@ def tracks_exist(s3_client, bucket, s3_path): return True +def capture_karaoke_config(shop, app_settings): + """Snapshot every value process_karaoke_detached needs, into plain dicts. + + Called in the parent request while the shop ORM object is live. The + grandchild process receives only immutable values — no DB reads + required until the final metadata write (after karaoke completes, + minutes later, by which time the parent has long committed). + """ + if shop.has_primary_s3: + s3_creds = { + "region": shop.primary_s3_region, + "endpoint": shop.primary_s3_endpoint, + "access_key": shop.primary_s3_access_key, + "secret_key": shop.primary_s3_secret_key, + "bucket": shop.primary_s3_bucket, + } + else: + s3_creds = { + "region": app_settings["bucket.secure_uploads.region"], + "endpoint": app_settings["bucket.secure_uploads.post_endpoint"], + "access_key": app_settings["bucket.secure_uploads.access_key"], + "secret_key": app_settings["bucket.secure_uploads.secret_key"], + "bucket": app_settings["bucket.secure_uploads"], + } + + mirror_creds = None + if shop.has_s3_mirror: + mirror_creds = { + "region": shop.mirror_s3_region, + "endpoint": shop.mirror_s3_endpoint, + "access_key": shop.mirror_s3_access_key, + "secret_key": shop.mirror_s3_secret_key, + "bucket": shop.mirror_s3_bucket, + } + + return { + "unsandbox_pk": shop.unsandbox_public_key, + "unsandbox_sk": shop.unsandbox_secret_key, + "s3": s3_creds, + "mirror": mirror_creds, + } + + +def process_karaoke_detached(product_id, file_key, extension, s3_path, + karaoke_config, db_url): + """Fire-and-forget karaoke processing via detached child process. + + Double-forks so the worker survives uWSGI recycling, then runs + process_karaoke on the given source file and updates the product's + file_metadata + file_bytes with the resulting instrumentals/vocals. + + Per-product lockfile (/tmp/karaoke_{product_id}.lock) prevents + concurrent runs for the same product — a second call while one is + in flight silently returns. + + All shop config (unsandbox keys, S3 creds, mirror creds) is + captured by the caller via capture_karaoke_config() before the + fork, so the grandchild never reads uncommitted data across the + fork boundary. The grandchild's only DB contact is a write at the + end — after karaoke completes (minutes), by which time the parent + request has long since committed. + + Returns immediately in the parent. All errors in the grandchild + are logged, never raised. + """ + from ..models.product import get_media_type + + lockfile = f"/tmp/karaoke_{product_id}.lock" + + # Pre-check in parent: skip fork if another karaoke is already running + try: + check_fd = open(lockfile, "w") + fcntl.flock(check_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + fcntl.flock(check_fd, fcntl.LOCK_UN) + check_fd.close() + except (IOError, OSError): + return + + pid = os.fork() + if pid > 0: + os.waitpid(pid, 0) + return + + # Intermediate child: detach from uWSGI + os.setsid() + pid2 = os.fork() + if pid2 > 0: + os._exit(0) + + # --- Grandchild: fully detached --- + import resource + maxfd = resource.getrlimit(resource.RLIMIT_NOFILE)[1] + if maxfd == resource.RLIM_INFINITY: + maxfd = 1024 + for fd in range(3, maxfd): + try: + os.close(fd) + except OSError: + pass + + lock_fd = None + try: + lock_fd = open(lockfile, "w") + fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + lock_fd.write(str(os.getpid())) + lock_fd.flush() + + import boto3 + from sqlalchemy import create_engine + from sqlalchemy.orm import Session as SASession + from ..models.product import Product + from .s3_mirror import _make_mirror_client, mirror_key + + is_video = (get_media_type(extension) == "video") + s3_key = f"{s3_path}/{file_key}" + + s3_cfg = karaoke_config["s3"] + bucket = s3_cfg["bucket"] + s3 = boto3.session.Session().client( + "s3", + region_name=s3_cfg["region"], + endpoint_url=s3_cfg["endpoint"], + aws_access_key_id=s3_cfg["access_key"], + aws_secret_access_key=s3_cfg["secret_key"], + ) + + log.info("Detached karaoke: product=%s key=%s", product_id, s3_key) + + sizes = process_karaoke( + s3, bucket, s3_key, s3_path, is_video, extension, + public_key=karaoke_config["unsandbox_pk"], + secret_key=karaoke_config["unsandbox_sk"], + ) + + if not sizes: + log.warning("Detached karaoke produced no tracks: product=%s", product_id) + return + + # Persist metadata — parent request has long since committed by now + engine = create_engine(db_url) + session = SASession(bind=engine) + try: + product = session.get(Product, product_id) + if product: + track_ext = extension if is_video else "wav" + for track_name in ("instrumentals", "vocals"): + product.set_file_metadata( + track_name, track_ext, f"{track_name}.{track_ext}" + ) + tmp = product.file_bytes + tmp.update(sizes) + product.file_bytes = tmp + session.add(product) + session.commit() + product.update_s3_acls(s3, bucket) + log.info("Detached karaoke done: product=%s inst=%dB vox=%dB", + product_id, sizes["instrumentals"], sizes["vocals"]) + except Exception: + session.rollback() + log.exception("Detached karaoke DB write failed: product=%s", product_id) + finally: + session.close() + engine.dispose() + + # Mirror synchronously — a daemon thread would die at os._exit + mirror_cfg = karaoke_config["mirror"] + if mirror_cfg: + dst = _make_mirror_client( + mirror_cfg["endpoint"], + mirror_cfg["region"], + mirror_cfg["access_key"], + mirror_cfg["secret_key"], + ) + for track_name in ("instrumentals", "vocals"): + mirror_key( + s3, bucket, f"{s3_path}/{track_name}", + dst, mirror_cfg["bucket"], + ) + except (IOError, OSError): + pass # another karaoke acquired lock between pre-check and here + except Exception: + log.exception("Detached karaoke grandchild failed: product=%s", product_id) + finally: + if lock_fd: + try: + lock_fd.close() + os.unlink(lockfile) + except OSError: + pass + os._exit(0) + + def backfill_karaoke_async(shop_id, session_factory, app_settings): """Backfill vocal isolation tracks for a shop's catalog. diff --git a/make_post_sell/views/product.py b/make_post_sell/views/product.py index 61ee26d..0238a48 100644 --- a/make_post_sell/views/product.py +++ b/make_post_sell/views/product.py @@ -450,42 +450,23 @@ def product_edit(request): session_factory=request.registry["dbsession_factory"], ) - # Generate vocal isolation tracks if this is audio/video + # Generate vocal isolation tracks if this is audio/video. + # Runs in a detached child — karaoke takes minutes and would hang + # the response (black screen after upload) if run inline. from ..models.product import get_media_type - from ..lib.karaoke import process_karaoke + from ..lib.karaoke import capture_karaoke_config, process_karaoke_detached upload_media_type = get_media_type(product.extensions.get(file_key)) shop = product.shop if upload_media_type in ("video", "audio") and shop.unsandbox_public_key and shop.unsandbox_secret_key: - is_video = (upload_media_type == "video") ext = product.extensions.get(file_key) - sizes = process_karaoke( - request.shop_uploads_client, - request.shop_bucket_name, - f"{product.s3_path}/{file_key}", - product.s3_path, is_video, ext, - public_key=shop.unsandbox_public_key, - secret_key=shop.unsandbox_secret_key, + process_karaoke_detached( + product_id=product.id, + file_key=file_key, + extension=ext, + s3_path=product.s3_path, + karaoke_config=capture_karaoke_config(shop, request.registry.settings), + db_url=str(request.dbsession.get_bind().url), ) - if sizes: - track_ext = ext if is_video else "wav" - for track_name in ("instrumentals", "vocals"): - product.set_file_metadata(track_name, track_ext, f"{track_name}.{track_ext}") - tmp = product.file_bytes - tmp.update(sizes) - product.file_bytes = tmp - request.dbsession.add(product) - request.dbsession.flush() - product.update_s3_acls(request.shop_uploads_client, request.shop_bucket_name) - - # Mirror karaoke tracks to shop's custom bucket - if shop.has_s3_mirror: - from ..lib.s3_mirror import mirror_keys_async - mirror_keys_async( - request.shop_uploads_client, - request.shop_bucket_name, - [f"{product.s3_path}/instrumentals", f"{product.s3_path}/vocals"], - shop, - ) # redirect back to this page to clear # the params posted by the s3 webhooks. diff --git a/make_post_sell/views/watch.py b/make_post_sell/views/watch.py index c4bf3ce..a5bfdf3 100644 --- a/make_post_sell/views/watch.py +++ b/make_post_sell/views/watch.py @@ -1,6 +1,4 @@ -import fcntl import logging -import os from pyramid.view import view_config @@ -245,9 +243,9 @@ def watch_json(request): def karaoke_process(request): """On-demand karaoke processing for a single product. - Forks a detached child process (survives uWSGI recycling) that runs - process_karaoke and updates the DB. Returns immediately with status. - The watch_json 10s refresh loop picks up the new URLs when done. + Validates the request, then hands off to process_karaoke_detached + which double-forks and runs karaoke in a detached child. The + watch_json 10s refresh loop picks up the new URLs when done. """ product_id = request.matchdict.get("product_id") from ..models.product import Product @@ -266,11 +264,7 @@ def karaoke_process(request): request.response.status_int = 400 return {"error": "Karaoke not configured for this shop"} - # Determine which file to process - if product.is_sellable: - file_key = "preview" - else: - file_key = "product" + file_key = "preview" if product.is_sellable else "product" extension = product.extensions.get(file_key) if not extension: @@ -286,138 +280,16 @@ def karaoke_process(request): if "instrumentals" in product.extensions and "vocals" in product.extensions: return {"status": "ready"} - # One-at-a-time guard per product - lockfile = f"/tmp/karaoke_{product_id}.lock" - try: - check_fd = open(lockfile, "w") - fcntl.flock(check_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) - fcntl.flock(check_fd, fcntl.LOCK_UN) - check_fd.close() - except (IOError, OSError): - return {"status": "processing"} - - # Capture values before fork - db_url = str(request.dbsession.get_bind().url) - s3_path = product.s3_path - s3_key = f"{s3_path}/{file_key}" - is_video = media_type == "video" - pk = shop.unsandbox_public_key - sk = shop.unsandbox_secret_key - app_settings = request.registry.settings - has_mirror = shop.has_s3_mirror - shop_id = shop.id - - # BYOB credentials - if shop.has_primary_s3: - s3_region = shop.primary_s3_region - s3_endpoint = shop.primary_s3_endpoint - s3_access = shop.primary_s3_access_key - s3_secret = shop.primary_s3_secret_key - bucket = shop.primary_s3_bucket - else: - s3_region = app_settings["bucket.secure_uploads.region"] - s3_endpoint = app_settings["bucket.secure_uploads.post_endpoint"] - s3_access = app_settings["bucket.secure_uploads.access_key"] - s3_secret = app_settings["bucket.secure_uploads.secret_key"] - bucket = app_settings["bucket.secure_uploads"] - - pid = os.fork() - if pid > 0: - os.waitpid(pid, 0) - return {"status": "processing"} - - # Intermediate child: detach from uWSGI - os.setsid() - pid2 = os.fork() - if pid2 > 0: - os._exit(0) - - # --- Grandchild: fully detached --- - import resource - maxfd = resource.getrlimit(resource.RLIMIT_NOFILE)[1] - if maxfd == resource.RLIM_INFINITY: - maxfd = 1024 - for fd in range(3, maxfd): - try: - os.close(fd) - except OSError: - pass - - lock_fd = None - try: - lock_fd = open(lockfile, "w") - fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) - lock_fd.write(str(os.getpid())) - lock_fd.flush() - - import boto3 - from sqlalchemy import create_engine - from sqlalchemy.orm import Session as SASession - from ..lib.karaoke import process_karaoke - - engine = create_engine(db_url) - session = SASession(bind=engine) - - s3 = boto3.session.Session().client( - "s3", - region_name=s3_region, - endpoint_url=s3_endpoint, - aws_access_key_id=s3_access, - aws_secret_access_key=s3_secret, - ) - - log.info("On-demand karaoke: product=%s key=%s", product_id, s3_key) - - sizes = process_karaoke( - s3, bucket, s3_key, s3_path, is_video, extension, - public_key=pk, secret_key=sk, - ) - - if sizes: - product = session.get(Product, product_id) - if product: - track_ext = extension if is_video else "wav" - for track_name in ("instrumentals", "vocals"): - product.set_file_metadata( - track_name, track_ext, f"{track_name}.{track_ext}" - ) - tmp = product.file_bytes - tmp.update(sizes) - product.file_bytes = tmp - session.add(product) - session.commit() - product.update_s3_acls(s3, bucket) - log.info("On-demand karaoke done: product=%s inst=%dB vox=%dB", - product_id, sizes["instrumentals"], sizes["vocals"]) - - if has_mirror: - from ..models.shop import Shop as ShopModel - shop_obj = session.get(ShopModel, shop_id) - if shop_obj: - from ..lib.s3_mirror import mirror_keys_async - mirror_keys_async( - s3, bucket, - [f"{s3_path}/instrumentals", f"{s3_path}/vocals"], - shop_obj, - ) - else: - log.warning("On-demand karaoke failed: product=%s", product_id) - - session.close() - engine.dispose() - - except (IOError, OSError): - pass - except Exception: - log.exception("On-demand karaoke child failed: product=%s", product_id) - finally: - if lock_fd: - try: - lock_fd.close() - os.unlink(lockfile) - except OSError: - pass - os._exit(0) + from ..lib.karaoke import capture_karaoke_config, process_karaoke_detached + process_karaoke_detached( + product_id=product_id, + file_key=file_key, + extension=extension, + s3_path=product.s3_path, + karaoke_config=capture_karaoke_config(shop, request.registry.settings), + db_url=str(request.dbsession.get_bind().url), + ) + return {"status": "processing"} @view_config(route_name="discovery_ring_json", renderer="json")