engineering/make_post_sell
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases 2 Packages Wiki Activity Actions

fix: run karaoke in detached child so upload response returns

Browse source 
Upload handler called process_karaoke synchronously — for audio/video
products, vocal isolation (download, ffmpeg, voxsplit, reupload) can
take several minutes. The response hung until karaoke finished, so
users saw a black screen after a successful upload while Caddy/uwsgi
timed out. The file itself was already in S3+DB, which is why hitting
Back showed the edit page with the upload present.

Fix: extract the fork+detach logic the on-demand path already used
(views/watch.py:karaoke_process) into a reusable helper,
process_karaoke_detached, in lib/karaoke.py. Both the upload path and
the on-demand path now hand off to the helper and return immediately.

The helper takes a snapshot of shop+S3+mirror config in the parent
(capture_karaoke_config) so the grandchild only touches the DB at the
end for the final metadata write — by then the parent has long since
committed. Mirror writes run synchronously in the grandchild instead
of via mirror_keys_async daemon threads that were dying at os._exit.
This commit is contained in:
russell@unturf.com 2026-04-16 15:05:20 -04:00
parent d757cbecbe
commit 8fcb4da92c
3 changed files with 217 additions and 172 deletions
Download patch file Download diff file Expand all files Collapse all files

192
make_post_sell/lib/karaoke.py
View file

@ -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.

41
make_post_sell/views/product.py
View file

@ -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.

156
make_post_sell/views/watch.py
View file

@ -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")