replace base miner worker with docker-compose setup (#149)

.gitignore

@@ -11,7 +11,6 @@ note.txt
 *.mp3
 *.js
 *.xml 
-*.txt
 .DS_Store
 *.json
 .env

docs/miner_setup.md

@@ -1,6 +1,12 @@
 # Running Miner
 
-A high-performance decentralized video processing miner that leverages **Video2X** for AI-powered video upscaling. This guide provides detailed instructions to set up, configure, and run the miner effectively
+A high-performance decentralized video processing miner that forwards validator requests to containerized video services. The current miner framework supports:
+
+- **Video2X upscaling** through the `upscaling-video2x` Docker Compose profile.
+- **FFmpeg upscaling** through the `upscaling-ffmpeg` Docker Compose profile.
+- **FFmpeg compression** through the `compression` Docker Compose service.
+
+`neurons/miner.py` no longer needs the old PM2-managed `services/upscaling/server.py` or `services/compress/server.py` workers. It forwards directly to the service container ports.
 
 ---
 
@@ -10,264 +16,250 @@ To achieve optimal results, we recommend the following setup:
 
 - **Operating System**: Ubuntu 24.04 LTS or higher  
   [Learn more about Ubuntu 24.04 LTS](https://ubuntu.com/blog/tag/ubuntu-24-04-lts)
-- **GPU**: NVIDIA RTX 4090 or higher  
-  (Required for efficient video upscaling using Video2X)
+- **GPU**: NVIDIA RTX 4090 or higher
+- **VRAM**: At least 16 GB per GPU
+- **Docker** and **Docker Compose**
+- **NVIDIA drivers** and `nvidia-container-toolkit`
+- **Python**: Version 3.10 or higher for running the miner process
 
 ---
 
-## Install PM2 (Process Manager)
+## Install System Dependencies
 
-**PM2** is used to manage and monitor the miner process. If you haven’t installed PM2 yet, follow these steps:
+Install common host dependencies:
 
-1. Install `npm` and PM2:
-   ```bash
-   sudo apt update
-   sudo apt install npm -y
-   sudo npm install pm2 -g
-   pm2 update
-   ```
+```bash
+sudo apt update
+sudo apt install -y git curl npm python3 python3-venv python3-pip
+```
 
-2. For more details, refer to the [PM2 Documentation](https://pm2.io/docs/runtime/guide/installation/).
+Install Docker and Docker Compose using the official Docker instructions for your host, then install NVIDIA Container Toolkit so Compose services can access the GPU.
 
----
+Verify Docker can see the GPU:
 
-## Install Redis
+```bash
+docker run --rm --gpus all nvidia/cuda:12.4.1-base-ubuntu22.04 nvidia-smi
+```
 
-1. Install 'redis'
-   ```bash
-   sudo apt update
-   sudo apt install redis-server
-   sudo systemctl start redis
-   sudo systemctl enable redis-server
-   sudo systemctl status redis
-   ```
+---
 
-## Install Project Dependencies
+## Install PM2
 
-### Prerequisites
+PM2 is optional for the processing containers, but useful for managing the miner process itself.
 
-- **Python**: Version 3.10 or higher
-- **pip**: Python package manager
-- **virtualenv** (optional): For dependency isolation
+```bash
+sudo npm install pm2 -g
+pm2 update
+```
 
 ---
 
+## Install Project Dependencies
+
 ### 1. Clone the Repository
 
-Clone the project repository to your local machine:
 ```bash
 git clone https://github.com/vidaio-subnet/vidaio-subnet.git
 cd vidaio-subnet
 ```
 
----
-
-### 2. Set Up a Virtual Environment (Recommended)
+### 2. Set Up a Virtual Environment
 
-Create and activate a virtual environment to isolate project dependencies:
 ```bash
 python3 -m venv venv
-source venv/bin/activate  
+source venv/bin/activate
 ```
 
----
-
-### 3. Install the Package and Dependencies
+### 3. Install the Package
 
-Install the project and its dependencies using `pip`:
 ```bash
 pip install -e .
 ```
 
 ---
 
-### 4. Configure Environment Variables
-
-To configure environment variables, follow these steps:
+## Configure Storage
 
-1. Create a `.env` file in the project root directory by referencing the provided `.env.template` file:
-   ```bash
-   cp .env.template .env
-   ```
+The miner process downloads validator payload URLs, places local files in a shared work directory, forwards local paths to the processing containers, uploads processed outputs to S3-compatible storage, and returns presigned URLs. The processing containers run without egress.
 
-2. Set up a bucket in cloud storage. The base miner code utilizes MinIO to connect with cloud storage services, so you'll need to prepare your bucket using a platform that supports MinIO integration, such as Backblaze. Alternatively, you can modify the code to suit your specific requirements.
-3. Add the required variables to the `.env` file. For example:
-   ```env
-   BUCKET_NAME="S3 buckent name"
-   BUCKET_COMPATIBLE_ENDPOINT="S3 bucket endpoint"
-   BUTKET_COMPATIBLE_ACCESS_KEY="S3 bucket personal access key"
-   BUCKET_COMPATIBLE_SECRET_KEY="S3 bucket personal secret key"
-   PEXELS_API_KEY="Your Pexels account api key"
-   WANDB_API_KEY="Your WANDB account api key"
-   ```
+Create a miner service environment file:
 
-4. Ensure that the bucket is configured with the appropriate permissions to allow file uploads and enable public access for downloads via presigned URLs.
-
-5. Once the `.env` file is properly configured, the application will use the specified credentials for S3 bucket and Pexels.
-
-
----
+```bash
+cp miner/.env.template miner/.env
+```
 
-## Install Video2X
+Edit `miner/.env` with your storage credentials:
+
+```env
+COMPOSE_PROJECT_NAME=miner
+MINER_SHARED_DIR=/tmp/vidaio-miner-video-tmp
+MINER_STORAGE_PROVIDER=backblaze
+MINER_STORAGE_S3_ACCESS_KEY_ID=your-access-key
+MINER_STORAGE_S3_SECRET_ACCESS_KEY=your-secret-key
+MINER_STORAGE_S3_REGION=us-east-005
+MINER_STORAGE_S3_BUCKET_NAME=your-bucket-name
+MINER_STORAGE_S3_ENDPOINT_URL=https://s3.us-east-005.backblazeb2.com
+```
 
-The miner requires **Video2X** for AI-powered video upscaling. Follow the steps below to install and configure Video2X.
+For AWS S3, use the same variable names and leave `MINER_STORAGE_S3_ENDPOINT_URL` blank.
 
----
+Create the shared work directory before starting the miner and containers:
 
-### Step 1: Install FFMPEG
-
-FFMPEG is required for processing video files. Install it using the following commands:
 ```bash
-sudo apt update
-sudo apt install ffmpeg -y
+mkdir -p /tmp/vidaio-miner-video-tmp
+chmod 777 /tmp/vidaio-miner-video-tmp
 ```
 
-For more details, refer to the [FFMPEG Documentation](https://www.ffmpeg.org/download.html#build-linux).
-
 ---
 
-### Step 2: Install CUDA and NVCC
+## Start Processing Services
 
-Ensure your CUDA drivers and NVCC (NVIDIA Compiler) are properly installed and configured to support GPU acceleration.
+Run Docker Compose commands from the `miner` directory:
 
-1. Verify your CUDA installation:
-   ```bash
-   nvcc --version
-   ```
+```bash
+cd miner
+```
 
-2. Ensure you CUDA driver is installed correctly
-   <!-- Install or update the CUDA drivers if they are not already installed:
-   ```bash
-   sudo apt update
-   sudo apt install nvidia-cuda-toolkit -y
-   ``` -->
+### Option A: Video2X Upscaling Miner
 
-For more information, refer to the [CUDA Toolkit Installation Guide](https://developer.nvidia.com/cuda-toolkit).
+Use this profile if you want Video2X-based upscaling. Complete the host/container setup first:
 
----
+```bash
+cat upscaling/SETUP.md 
+```
 
-### Step 3: Install Video2X
+After running all the commands listed above, start the Video2X upscaling service:
 
----
+```bash
+docker compose --profile upscaling-video2x up -d upscaling-video2x
+```
 
-### Option 1: Install Video2X by downloading Debian Package:
+The service is exposed on `http://localhost:8003`.
 
+### Option B: FFmpeg Upscaling Miner
 
-1. **Download the Video2X `.deb` package**:
-   ```bash
-   wget -P services/upscaling/models https://github.com/k4yt3x/video2x/releases/download/6.3.1/video2x-linux-ubuntu2404-amd64.deb
-   ```
+Use this profile if you want FFmpeg-based upscaling instead of Video2X:
 
-2. **Install the package using `dpkg`**:
-   ```bash
-   sudo dpkg -i services/upscaling/models/video2x-linux-ubuntu2404-amd64.deb
-   ```
+```bash
+docker compose --profile upscaling-ffmpeg up -d upscaling-ffmpeg
+```
 
-3. **Resolve dependencies** (if any):
-   ```bash
-   sudo apt-get install -f
-   ```
+The FFmpeg upscaling service is exposed on `http://localhost:8005`.
 
-### Option 2: Install Video2X by Building Debian Package:
+Before starting `neurons/miner.py`, point the miner at that port:
 
-1. Install Cargo
-Cargo is required to build the Video2X package:
 ```bash
-sudo apt-get update
-sudo apt-get install cargo -y
-cargo install just --version=1.39.0
+export MINER_UPSCALING_SERVICE_URL=http://localhost:8005
 ```
 
-2. Clone the Video2X Repository
-you can clone this repository within the current vidaio-subnet package
-```bash
-git clone --recurse-submodules https://github.com/vidAio-subnet/video2x
-cd video2x
-```
+### Option C: Compression Miner
 
-3. Build the Video2X Project
-Before building, ensure `~/.cargo/bin` is included in your `PATH` environment variable:
-```bash
-export PATH="$HOME/.cargo/bin:$PATH"
-```
+Start the compression service:
 
-Run the following command to build the package:
 ```bash
-just ubuntu2404
+docker compose up -d compression
 ```
 
-Once the build is complete, the `.deb` package will be located in the current directory.
+The compression service is exposed on `http://localhost:8004`.
+
+Return to the repository root before launching the miner:
 
-4. Install the Built Package
-Install the `.deb` package using:
 ```bash
-sudo dpkg -i video2x-linux-ubuntu-amd64.deb
+cd ..
 ```
 
 ---
-For additional details, refer to the [Video2X Documentation](https://docs.video2x.org/building/linux.html).
 
-## Running the Video Upscaling Endpoint
+## Miner Forwarding Configuration
 
-You can run the video upscaling endpoint using **PM2** to manage the process:
+`neurons/miner.py` forwards requests to these service URLs:
 
 ```bash
-pm2 start "python services/upscaling/server.py" --name video-upscaler
+export MINER_UPSCALING_SERVICE_URL=http://localhost:8003
+export MINER_COMPRESSION_SERVICE_URL=http://localhost:8004
+export MINER_SHARED_VOLUME_PATH=/tmp/vidaio-miner-video-tmp
 ```
 
-### Notes:
-- The `video-upscaler` process will handle video upscaling requests.
-- Use the following PM2 commands to manage the process:
-  - **View Logs**: `pm2 logs video-upscaler`
-  - **Restart**: `pm2 restart video-upscaler`
-  - **Stop**: `pm2 stop video-upscaler`
+Defaults:
 
----
-
-## Running the Video Compression Endpoint
+- `MINER_UPSCALING_SERVICE_URL` defaults to `http://localhost:8003` for Video2X upscaling.
+- `MINER_COMPRESSION_SERVICE_URL` defaults to `http://localhost:8004`.
+- `MINER_SHARED_VOLUME_PATH` defaults to `MINER_SHARED_DIR`, then `/tmp/vidaio-miner-video-tmp`.
+- For FFmpeg upscaling, override `MINER_UPSCALING_SERVICE_URL` to `http://localhost:8005`.
 
-You can also run the video Compression endpoint using **PM2** to manage the process:
+Useful service checks:
 
 ```bash
-pm2 start "python services/compress/server.py" --name video-compressor
+curl -sf http://localhost:8003/health
+curl -sf http://localhost:8004/health
+curl -sf http://localhost:8005/health
 ```
 
+Only the services you started need to pass health checks.
+
 ---
 
-## Running the file deletion process
+## Run the Miner
 
-You can run the file deletion process using **PM2** to manage the process:
+Start the miner from the repository root:
 
 ```bash
-pm2 start "python services/miner_utilities/file_deletion_server.py" --name video-deleter
+python3 neurons/miner.py \
+  --wallet.name [Your_Wallet_Name] \
+  --wallet.hotkey [Your_Hotkey_Name] \
+  --subtensor.network finney \
+  --netuid 85 \
+  --axon.port [port] \
+  --logging.debug
 ```
 
-## Running the Miner with PM2
+### Run the Miner with PM2
 
-To run the miner, use the following command:
+Export any `MINER_*_SERVICE_URL` overrides in the same shell before starting PM2. If you change those variables later, restart with `--update-env`.
 
 ```bash
 pm2 start "python3 neurons/miner.py --wallet.name [Your_Wallet_Name] --wallet.hotkey [Your_Hotkey_Name] --subtensor.network finney --netuid 85 --axon.port [port] --logging.debug" --name video-miner
 ```
 
-### Parameters:
-- **`--wallet.name`**: Replace `[Your_Wallet_Name]` with your wallet name.
-- **`--wallet.hotkey`**: Replace `[Your_Hotkey_Name]` with your hotkey name.
-- **`--subtensor.network`**: Specify the target network (e.g., `finney`).
-- **`--netuid`**: Specify the network UID (e.g., `292`).
-- **`--axon.port`**: Replace `[port]` with the desired port number.
-- **`--logging.debug`**: Enables debug-level logging for detailed output.
+### Parameters
+
+- `--wallet.name`: Replace `[Your_Wallet_Name]` with your wallet name.
+- `--wallet.hotkey`: Replace `[Your_Hotkey_Name]` with your hotkey name.
+- `--subtensor.network`: Specify the target network, usually `finney`.
+- `--netuid`: Specify the network UID, currently `85`.
+- `--axon.port`: Replace `[port]` with the desired public axon port.
+- `--logging.debug`: Enables debug-level logging.
+
+---
+
+## Managing Services
+
+Processing containers:
+
+```bash
+cd miner
+docker compose ps
+docker compose logs -f upscaling-video2x
+docker compose logs -f upscaling-ffmpeg
+docker compose logs -f compression
+docker compose down
+```
+
+Miner process:
 
-### Managing the Miner Process:
-- **Start the Miner**: The above command will start the miner as a PM2 process named `video-miner`.
-- **View Logs**: Use `pm2 logs video-miner` to monitor miner logs in real time.
-- **Restart the Miner**: Use `pm2 restart video-miner` to restart the process.
-- **Stop the Miner**: Use `pm2 stop video-miner` to stop the process.
+```bash
+pm2 logs video-miner
+pm2 restart video-miner --update-env
+pm2 stop video-miner
+```
 
 ---
 
 ## Additional Notes
 
-- Ensure all dependencies are installed and configured correctly before running the miner.
-- Use a high-performance GPU and sufficient system resources for optimal performance.
-- For troubleshooting and debugging, refer to the logs available in PM2.
+- Run either `upscaling-video2x` or `upscaling-ffmpeg` for an upscaling miner. They expose different host ports but both implement `/upscale`.
+- Run `compression` for a compression miner. It implements `/compress`.
+- Processing containers publish localhost-only service ports and have `DISABLE_REMOTE_IO=true`; they only process local files from the shared work directory by default.
+- The old PM2 service scripts for upscaling and compression are redundant with this container framework.
+- For Video2X-specific preparation, use `miner/upscaling/SETUP.md`.
+- For FFmpeg upscaling build/runtime notes, use `miner/upscaling/ffmpeg/SETUP.md`.

miner/.env.template

@@ -0,0 +1,15 @@
+COMPOSE_PROJECT_NAME=miner
+MINER_SHARED_DIR=/tmp/vidaio-miner-video-tmp
+
+# S3-compatible storage settings
+MINER_STORAGE_PROVIDER=backblaze
+MINER_STORAGE_S3_ACCESS_KEY_ID=your-access-key
+MINER_STORAGE_S3_SECRET_ACCESS_KEY=your-secret-key
+MINER_STORAGE_S3_REGION=us-east-005
+MINER_STORAGE_S3_BUCKET_NAME=your-bucket-name
+MINER_STORAGE_S3_ENDPOINT_URL=https://s3.us-east-005.backblazeb2.com
+
+# For AWS S3, keep the same variable names and leave the endpoint blank.
+# MINER_STORAGE_S3_REGION=us-west-1
+# MINER_STORAGE_S3_BUCKET_NAME=vidaiosubnet
+# MINER_STORAGE_S3_ENDPOINT_URL=

miner/README.md

@@ -0,0 +1,53 @@
+# Miner Quickstart
+
+This project includes three miner services:
+- `upscaling-video2x`: Upscaling service that wraps Video2X.
+- `upscaling-ffmpeg`: Upscaling service using FFmpeg directly.
+- `compression`: Compression microservice.
+
+## Prerequisites
+- Run `sudo apt update`
+- Docker + Docker Compose
+- NVIDIA drivers and `nvidia-container-toolkit`
+- Storage environment variables (S3-compatible) in a new `miner/.env` file derived from `miner/.env.template`
+- A shared local work directory. The default is `/tmp/vidaio-miner-video-tmp` via `MINER_SHARED_DIR`.
+
+The service APIs are published on localhost-only Docker ports. `neurons/miner.py` downloads validator payloads, places local files in `MINER_SHARED_DIR`, forwards local paths to the containers, uploads processed outputs, and returns presigned URLs. Remote URL input is disabled inside the service containers by default.
+
+```bash
+mkdir -p /tmp/vidaio-miner-video-tmp
+chmod 777 /tmp/vidaio-miner-video-tmp
+```
+
+## Video2X Upscaling Miner
+1. Complete host/container preparation in `miner/upscaling/SETUP.md`.
+2. Start service:
+   ```bash
+   docker compose --profile upscaling-video2x up -d upscaling-video2x
+   ```
+3. Point the miner at the Video2X profile port:
+   ```bash
+   export MINER_UPSCALING_SERVICE_URL=http://localhost:8003
+   ```
+
+## FFmpeg Upscaling Miner
+1. (Optional) Review build/runtime notes in `miner/upscaling/ffmpeg/SETUP.md`.
+2. Start service:
+   ```bash
+   docker compose --profile upscaling-ffmpeg up -d upscaling-ffmpeg
+   ```
+3. Point the miner at the FFmpeg profile port:
+   ```bash
+   export MINER_UPSCALING_SERVICE_URL=http://localhost:8005
+   ```
+
+## Compression Miner
+```bash
+docker compose up -d compression
+```
+
+## Miner Forwarding
+`neurons/miner.py` forwards directly to the container services:
+- `MINER_UPSCALING_SERVICE_URL` defaults to `http://localhost:8003`.
+- `MINER_COMPRESSION_SERVICE_URL` defaults to `http://localhost:8004`.
+- `MINER_SHARED_VOLUME_PATH` defaults to `MINER_SHARED_DIR`, then `/tmp/vidaio-miner-video-tmp`.

miner/compression/Dockerfile

@@ -0,0 +1,68 @@
+FROM nvidia/cuda:13.0.0-devel-ubuntu22.04
+
+# Build dependencies for ffmpeg + NVENC/NVDEC + codec libraries
+RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
+    build-essential ninja-build nasm perl g++ cmake wget unzip \
+    nvidia-cuda-dev nvidia-cuda-toolkit pkg-config \
+    libx264-dev libx265-dev libvpx-dev \
+    git python3-pip python3-venv curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install meson (needed for dav1d build)
+RUN pip3 install --no-cache-dir meson
+
+# nv-codec-headers (NVENC/NVDEC API headers)
+RUN wget https://github.com/FFmpeg/nv-codec-headers/archive/876af32a202d0de83bd1d36fe74ee0f7fcf86b0d.zip \
+    && unzip 876af32a202d0de83bd1d36fe74ee0f7fcf86b0d.zip \
+    && cd nv-codec-headers-876af32a202d0de83bd1d36fe74ee0f7fcf86b0d \
+    && make && make install && cd .. \
+    && rm -rf nv-codec-headers-* 876af32a202d0de83bd1d36fe74ee0f7fcf86b0d.zip
+
+# dav1d (AV1 decoder)
+RUN git clone --depth 1 --branch 1.4.2 https://code.videolan.org/videolan/dav1d.git \
+    && cd dav1d && mkdir build && cd build \
+    && meson setup .. --default-library=static --buildtype release \
+    && ninja && ninja install && ldconfig \
+    && cd ../.. && rm -rf dav1d
+
+# SVT-AV1 (AV1 encoder for libsvtav1)
+RUN git clone --depth 1 --branch v4.1.0 https://gitlab.com/AOMediaCodec/SVT-AV1.git \
+    && cd SVT-AV1 && mkdir build && cd build \
+    && cmake .. -G"Unix Makefiles" -DCMAKE_BUILD_TYPE=Release -DBUILD_SHARED_LIBS=ON \
+    && make -j$(nproc) && make install && ldconfig \
+    && cd ../.. && rm -rf SVT-AV1
+
+# Build ffmpeg (pinned commit 33b215d) with CUDA + x264 + x265 + dav1d + svtav1
+RUN export PKG_CONFIG_PATH="/usr/local/lib/pkgconfig:/usr/local/lib/x86_64-linux-gnu/pkgconfig:$PKG_CONFIG_PATH" \
+    && git init ffmpeg && cd ffmpeg \
+    && git remote add origin https://github.com/FFmpeg/FFmpeg.git \
+    && git fetch --depth 1 origin 33b215d1554a14e87416a24f8e6034312e629af7 \
+    && git checkout FETCH_HEAD \
+    && ./configure \
+        --enable-gpl \
+        --enable-libx264 \
+        --enable-libx265 \
+        --enable-libvpx \
+        --enable-libdav1d \
+        --enable-libsvtav1 \
+        --enable-nonfree \
+        --enable-cuda-nvcc \
+        --enable-nvenc \
+        --enable-nvdec \
+        --extra-cflags="-I/usr/local/cuda/include" \
+        --extra-ldflags="-L/usr/local/cuda/lib64" \
+        --disable-static \
+        --enable-shared \
+        --nvccflags="-gencode arch=compute_75,code=sm_75 -gencode arch=compute_75,code=compute_75 -O2" \
+    && make -j$(nproc) && make install && ldconfig \
+    && cd .. && rm -rf ffmpeg
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY app.py .
+
+EXPOSE 8004
+CMD ["python3", "-m", "uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8004"]

miner/compression/app.py

@@ -0,0 +1,296 @@
+"""
+Compression microservice — wraps the ffmpeg binary installed in this container.
+
+Accepts a video path (on shared volume) OR a URL, codec, and quality settings,
+runs GPU-accelerated ffmpeg compression, returns the output path or S3 URL.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import uuid
+from typing import Optional
+
+import boto3
+import httpx
+from botocore.config import Config
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, Field
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s | %(levelname)-7s | %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+log = logging.getLogger("compression")
+
+app = FastAPI(title="Video Compression Service")
+
+SHARED_VOLUME_PATH = os.getenv("SHARED_VOLUME_PATH", "/tmp/organic-proxy")
+MAX_CONCURRENT = int(os.getenv("MAX_CONCURRENT_COMPRESSION", "2"))
+DISABLE_REMOTE_IO = os.getenv("DISABLE_REMOTE_IO", "false").lower() in ("1", "true", "yes")
+
+# Storage provider label only. Uploads use one S3-compatible code path.
+STORAGE_PROVIDER = os.getenv("MINER_STORAGE_PROVIDER", "s3").lower()
+S3_REGION = os.getenv("MINER_STORAGE_S3_REGION", "us-east-1").strip() or "us-east-1"
+S3_BUCKET = os.getenv("MINER_STORAGE_S3_BUCKET_NAME", "").strip()
+S3_ACCESS_KEY_ID = os.getenv("MINER_STORAGE_S3_ACCESS_KEY_ID", "").strip()
+S3_SECRET_ACCESS_KEY = os.getenv("MINER_STORAGE_S3_SECRET_ACCESS_KEY", "").strip()
+S3_ENDPOINT_URL = (
+    os.getenv("MINER_STORAGE_S3_ENDPOINT_URL")
+    or os.getenv("MINER_STORAGE_S3_ENDPOINT")
+    or ""
+).strip()
+S3_PRESIGNED_EXPIRY = int(
+    os.getenv("MINER_STORAGE_S3_PRESIGNED_EXPIRY")
+    or os.getenv("S3_PRESIGNED_EXPIRY")
+    or "3600"
+)
+
+_semaphore = asyncio.Semaphore(MAX_CONCURRENT)
+_queue_size = 0
+_active_count = 0
+_lock = asyncio.Lock()
+
+# Codec name → nvenc/software encoder mapping
+CODEC_MAP = {
+    "AV1": "av1_nvenc",
+    "H264": "h264_nvenc",
+    "H.264": "h264_nvenc",
+    "HEVC": "hevc_nvenc",
+    "H265": "hevc_nvenc",
+    "H.265": "hevc_nvenc",
+    "VP9": "libvpx-vp9",
+}
+
+
+def _is_url(path: str) -> bool:
+    return path.startswith("http://") or path.startswith("https://")
+
+
+def _get_s3_client():
+    client_kwargs = {
+        "region_name": S3_REGION,
+        "aws_access_key_id": S3_ACCESS_KEY_ID or None,
+        "aws_secret_access_key": S3_SECRET_ACCESS_KEY or None,
+        "config": Config(signature_version="s3v4"),
+    }
+    if S3_ENDPOINT_URL:
+        client_kwargs["endpoint_url"] = S3_ENDPOINT_URL
+    return boto3.client("s3", **client_kwargs)
+
+
+def _storage_config_status() -> dict[str, object]:
+    return {
+        "provider": STORAGE_PROVIDER,
+        "region": S3_REGION,
+        "bucket_configured": bool(S3_BUCKET),
+        "access_key_configured": bool(S3_ACCESS_KEY_ID),
+        "secret_key_configured": bool(S3_SECRET_ACCESS_KEY),
+        "endpoint_configured": bool(S3_ENDPOINT_URL),
+    }
+
+
+def _validate_s3_config():
+    missing = []
+    if not S3_BUCKET:
+        missing.append("MINER_STORAGE_S3_BUCKET_NAME")
+    if not S3_ACCESS_KEY_ID:
+        missing.append("MINER_STORAGE_S3_ACCESS_KEY_ID")
+    if not S3_SECRET_ACCESS_KEY:
+        missing.append("MINER_STORAGE_S3_SECRET_ACCESS_KEY")
+
+    if missing:
+        raise RuntimeError(f"Missing storage configuration: {', '.join(missing)}")
+
+
+async def _download_url(url: str, dest: str):
+    log.info(f"Downloading {url[:80]}... → {dest}")
+    async with httpx.AsyncClient(timeout=600.0, follow_redirects=True) as client:
+        async with client.stream("GET", url) as resp:
+            resp.raise_for_status()
+            with open(dest, "wb") as f:
+                async for chunk in resp.aiter_bytes(chunk_size=8192):
+                    f.write(chunk)
+    log.info(f"Downloaded {os.path.getsize(dest) / (1024*1024):.1f} MB → {dest}")
+
+
+def _upload_to_s3(local_path: str, key: str) -> str:
+    _validate_s3_config()
+    client = _get_s3_client()
+    client.upload_file(local_path, S3_BUCKET, key)
+    url = client.generate_presigned_url(
+        "get_object",
+        Params={"Bucket": S3_BUCKET, "Key": key},
+        ExpiresIn=min(S3_PRESIGNED_EXPIRY, 604800),
+    )
+    log.info(f"Uploaded to {STORAGE_PROVIDER} storage: s3://{S3_BUCKET}/{key}")
+    return url
+
+
+def _cleanup(*paths: str):
+    for p in paths:
+        if p and os.path.exists(p):
+            try:
+                os.remove(p)
+            except OSError:
+                pass
+
+
+class CompressRequest(BaseModel):
+    video_path: str = Field(..., description="Local path or URL to input video")
+    task_id: str = Field("", description="Task ID for logging")
+    codec: str = Field("AV1", description="Target codec: AV1, H264, HEVC, VP9")
+    codec_mode: str = Field("CRF", description="Rate control mode: CRF or VBR")
+    cq: int = Field(35, description="Constant quality value (lower = higher quality)")
+    preset: str = Field("p4", description="Encoder preset")
+    target_bitrate: Optional[int] = Field(None, description="Target bitrate in bps (for VBR mode)")
+    target_width: Optional[int] = Field(None, description="Target width for downscaling")
+    target_height: Optional[int] = Field(None, description="Target height for downscaling")
+
+
+class CompressResponse(BaseModel):
+    output_path: str = Field("", description="Local path (local mode) or empty (remote mode)")
+    output_url: str = Field("", description="S3 presigned URL (remote mode) or empty (local mode)")
+    success: bool
+    error: Optional[str] = None
+    active_tasks: Optional[int] = None
+    queued_tasks: Optional[int] = None
+
+
+@app.get("/health")
+async def health():
+    return {
+        "status": "ok",
+        "max_concurrent": MAX_CONCURRENT,
+        "active_tasks": _active_count,
+        "queued_tasks": max(0, _queue_size - _active_count),
+        "storage": _storage_config_status(),
+    }
+
+
+@app.get("/queue")
+async def queue_status():
+    return {
+        "max_concurrent": MAX_CONCURRENT,
+        "active_tasks": _active_count,
+        "queued_tasks": max(0, _queue_size - _active_count),
+        "total_pending": _queue_size,
+    }
+
+
+@app.post("/compress", response_model=CompressResponse)
+async def compress(req: CompressRequest):
+    task_label = req.task_id or uuid.uuid4().hex[:8]
+    remote_mode = _is_url(req.video_path)
+    encoder = CODEC_MAP.get(req.codec.upper(), "av1_nvenc")
+
+    if remote_mode and DISABLE_REMOTE_IO:
+        return CompressResponse(success=False, error="Remote URL input is disabled for this local-only service")
+
+    os.makedirs(SHARED_VOLUME_PATH, exist_ok=True)
+
+    # --- Resolve input to local path ---
+    if remote_mode:
+        local_input = os.path.join(SHARED_VOLUME_PATH, f"{task_label}_input.mp4")
+        try:
+            await _download_url(req.video_path, local_input)
+        except Exception as e:
+            return CompressResponse(success=False, error=f"Failed to download input: {e}")
+    else:
+        local_input = req.video_path
+        if not os.path.exists(local_input):
+            raise HTTPException(status_code=400, detail=f"Input file not found: {local_input}")
+
+    basename = os.path.splitext(os.path.basename(local_input))[0]
+    output_filename = f"{basename}_compressed.mp4"
+    output_path = os.path.join(SHARED_VOLUME_PATH, output_filename)
+
+    # Build ffmpeg command
+    ffmpeg_args = [
+        "ffmpeg", "-y",
+        "-hwaccel", "cuda",
+        "-i", local_input,
+    ]
+
+    ffmpeg_args.extend(["-c:v", encoder])
+
+    if req.codec_mode == "VBR" and req.target_bitrate:
+        ffmpeg_args.extend(["-b:v", str(req.target_bitrate)])
+    else:
+        ffmpeg_args.extend(["-cq", str(req.cq)])
+
+    ffmpeg_args.extend(["-preset", req.preset])
+
+    if req.target_width and req.target_height:
+        w = req.target_width if req.target_width % 2 == 0 else req.target_width - 1
+        h = req.target_height if req.target_height % 2 == 0 else req.target_height - 1
+        ffmpeg_args.extend(["-vf", f"scale={w}:{h}"])
+
+    ffmpeg_args.extend(["-c:a", "copy"])
+    ffmpeg_args.append(output_path)
+
+    cmd = ffmpeg_args
+
+    global _queue_size, _active_count
+
+    async with _lock:
+        _queue_size += 1
+        queued = max(0, _queue_size - _active_count)
+
+    log.info(f"[{task_label}] Queued compression (codec={encoder}, cq={req.cq}, waiting={queued}, remote={remote_mode})")
+
+    try:
+        async with _semaphore:
+            async with _lock:
+                _active_count += 1
+                queued_now = max(0, _queue_size - _active_count)
+
+            log.info(f"[{task_label}] Starting compression (active={_active_count}/{MAX_CONCURRENT}, queued={queued_now})")
+            log.info(f"[{task_label}] Command: {' '.join(cmd)}")
+
+            proc = await asyncio.create_subprocess_exec(
+                *cmd,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            stdout, stderr = await proc.communicate()
+
+            async with _lock:
+                _active_count -= 1
+    finally:
+        async with _lock:
+            _queue_size -= 1
+
+    stats = dict(active_tasks=_active_count, queued_tasks=max(0, _queue_size - _active_count))
+
+    if proc.returncode != 0:
+        err_msg = stderr.decode(errors="replace").strip()
+        log.error(f"[{task_label}] ffmpeg failed (rc={proc.returncode}): {err_msg}")
+        if remote_mode:
+            _cleanup(local_input)
+        return CompressResponse(success=False, error=err_msg, **stats)
+
+    if not os.path.exists(output_path):
+        log.error(f"[{task_label}] Output file not found: {output_path}")
+        if remote_mode:
+            _cleanup(local_input)
+        return CompressResponse(success=False, error="Output file not created", **stats)
+
+    # --- Remote mode: upload result to S3, return URL ---
+    if remote_mode:
+        try:
+            s3_key = f"processing/{task_label}/{output_filename}"
+            output_url = _upload_to_s3(output_path, s3_key)
+            log.info(f"[{task_label}] Compression complete (remote): {output_url[:80]}...")
+            return CompressResponse(output_url=output_url, success=True, **stats)
+        except Exception as e:
+            log.error(f"[{task_label}] S3 upload failed: {e}")
+            return CompressResponse(success=False, error=f"S3 upload failed: {e}", **stats)
+        finally:
+            _cleanup(local_input, output_path)
+
+    log.info(f"[{task_label}] Compression complete (local): {output_path}")
+    return CompressResponse(output_path=output_path, success=True, **stats)

miner/compression/requirements.txt

@@ -0,0 +1,5 @@
+fastapi==0.115.12
+uvicorn==0.34.2
+pydantic==2.11.1
+boto3==1.35.99
+httpx==0.28.1

miner/docker-compose.yml

@@ -0,0 +1,106 @@
+x-upscaling-video2x-base: &upscaling-video2x-base
+  build: ./upscaling
+  restart: unless-stopped
+  profiles: ["upscaling-video2x"]
+  networks:
+    - miner-local
+  environment:
+    - VIDEO2X_IMAGE=ghcr.io/k4yt3x/video2x:6.4.0
+    - SHARED_VOLUME_PATH=/tmp/organic-proxy
+    - HOST_SHARED_VOLUME_PATH=${MINER_SHARED_DIR:-/tmp/vidaio-miner-video-tmp}
+    - DOCKER_VOLUME_NAME=${COMPOSE_PROJECT_NAME:-organic-proxy}_video-tmp
+    - DISABLE_REMOTE_IO=true
+    - MAX_CONCURRENT_UPSCALING=2
+    - MINER_STORAGE_PROVIDER=${MINER_STORAGE_PROVIDER:-s3}
+    - MINER_STORAGE_S3_REGION=${MINER_STORAGE_S3_REGION:-us-east-1}
+    - MINER_STORAGE_S3_BUCKET_NAME=${MINER_STORAGE_S3_BUCKET_NAME:-}
+    - MINER_STORAGE_S3_ACCESS_KEY_ID=${MINER_STORAGE_S3_ACCESS_KEY_ID:-}
+    - MINER_STORAGE_S3_SECRET_ACCESS_KEY=${MINER_STORAGE_S3_SECRET_ACCESS_KEY:-}
+    - MINER_STORAGE_S3_ENDPOINT_URL=${MINER_STORAGE_S3_ENDPOINT_URL:-${MINER_STORAGE_S3_ENDPOINT:-}}
+    - MINER_STORAGE_S3_PRESIGNED_EXPIRY=${MINER_STORAGE_S3_PRESIGNED_EXPIRY:-${S3_PRESIGNED_EXPIRY:-3600}}
+  volumes:
+    - ${MINER_SHARED_DIR:-/tmp/vidaio-miner-video-tmp}:/tmp/organic-proxy
+    - /var/run/docker.sock:/var/run/docker.sock
+  healthcheck:
+    test: ["CMD", "curl", "-sf", "http://localhost:8003/health"]
+    interval: 30s
+    timeout: 10s
+    retries: 3
+    start_period: 10s
+
+x-upscaling-ffmpeg-base: &upscaling-ffmpeg-base
+  build: ./upscaling/ffmpeg
+  restart: unless-stopped
+  profiles: ["upscaling-ffmpeg"]
+  gpus: all
+  networks:
+    - miner-local
+  environment:
+    - NVIDIA_VISIBLE_DEVICES=all
+    - NVIDIA_DRIVER_CAPABILITIES=video,compute,utility
+    - SHARED_VOLUME_PATH=/tmp/organic-proxy
+    - DISABLE_REMOTE_IO=true
+    - MAX_CONCURRENT_UPSCALING=2
+    - MINER_STORAGE_PROVIDER=${MINER_STORAGE_PROVIDER:-s3}
+    - MINER_STORAGE_S3_REGION=${MINER_STORAGE_S3_REGION:-us-east-1}
+    - MINER_STORAGE_S3_BUCKET_NAME=${MINER_STORAGE_S3_BUCKET_NAME:-}
+    - MINER_STORAGE_S3_ACCESS_KEY_ID=${MINER_STORAGE_S3_ACCESS_KEY_ID:-}
+    - MINER_STORAGE_S3_SECRET_ACCESS_KEY=${MINER_STORAGE_S3_SECRET_ACCESS_KEY:-}
+    - MINER_STORAGE_S3_ENDPOINT_URL=${MINER_STORAGE_S3_ENDPOINT_URL:-${MINER_STORAGE_S3_ENDPOINT:-}}
+    - MINER_STORAGE_S3_PRESIGNED_EXPIRY=${MINER_STORAGE_S3_PRESIGNED_EXPIRY:-${S3_PRESIGNED_EXPIRY:-3600}}
+  volumes:
+    - ${MINER_SHARED_DIR:-/tmp/vidaio-miner-video-tmp}:/tmp/organic-proxy
+  healthcheck:
+    test: ["CMD", "curl", "-sf", "http://localhost:8003/health"]
+    interval: 30s
+    timeout: 10s
+    retries: 3
+    start_period: 10s
+
+x-compression-base: &compression-base
+  build: ./compression
+  restart: unless-stopped
+  gpus: all
+  networks:
+    - miner-local
+  environment:
+    - NVIDIA_VISIBLE_DEVICES=all
+    - NVIDIA_DRIVER_CAPABILITIES=video,compute,utility
+    - SHARED_VOLUME_PATH=/tmp/organic-proxy
+    - DISABLE_REMOTE_IO=true
+    - MAX_CONCURRENT_COMPRESSION=2
+    - MINER_STORAGE_PROVIDER=${MINER_STORAGE_PROVIDER:-s3}
+    - MINER_STORAGE_S3_REGION=${MINER_STORAGE_S3_REGION:-us-east-1}
+    - MINER_STORAGE_S3_BUCKET_NAME=${MINER_STORAGE_S3_BUCKET_NAME:-}
+    - MINER_STORAGE_S3_ACCESS_KEY_ID=${MINER_STORAGE_S3_ACCESS_KEY_ID:-}
+    - MINER_STORAGE_S3_SECRET_ACCESS_KEY=${MINER_STORAGE_S3_SECRET_ACCESS_KEY:-}
+    - MINER_STORAGE_S3_ENDPOINT_URL=${MINER_STORAGE_S3_ENDPOINT_URL:-${MINER_STORAGE_S3_ENDPOINT:-}}
+    - MINER_STORAGE_S3_PRESIGNED_EXPIRY=${MINER_STORAGE_S3_PRESIGNED_EXPIRY:-${S3_PRESIGNED_EXPIRY:-3600}}
+  volumes:
+    - ${MINER_SHARED_DIR:-/tmp/vidaio-miner-video-tmp}:/tmp/organic-proxy
+  healthcheck:
+    test: ["CMD", "curl", "-sf", "http://localhost:8004/health"]
+    interval: 30s
+    timeout: 10s
+    retries: 3
+    start_period: 10s
+
+services:
+  upscaling-video2x:
+    <<: *upscaling-video2x-base
+    ports:
+      - "127.0.0.1:8003:8003"
+
+  upscaling-ffmpeg:
+    <<: *upscaling-ffmpeg-base
+    ports:
+      - "127.0.0.1:8005:8003"
+
+  compression:
+    <<: *compression-base
+    ports:
+      - "127.0.0.1:8004:8004"
+
+networks:
+  miner-local:
+    driver: bridge

miner/requirements.txt

@@ -0,0 +1,57 @@
+# Web Framework
+fastapi==0.110.3
+uvicorn==0.23.2
+pydantic==2.4.2
+# starlette==0.27.0
+
+# Video Processing
+ffmpeg-python==0.2.0
+moviepy==1.0.3
+
+# Storage
+boto3==1.28.64
+botocore==1.31.64
+minio
+
+# Task Management
+redis==5.0.1
+celery==5.3.4
+
+# HTTP Client
+httpx==0.25.0
+firerequests
+
+# Configuration
+pyyaml==6.0.1
+python-dotenv==1.0.0
+
+# Utilities
+# numpy==1.26.0
+pillow==10.0.1
+tqdm==4.66.1
+
+# Logging
+loguru==0.7.2
+python-json-logger
+
+
+# Testing
+pytest==7.4.2
+pytest-asyncio==0.21.1
+pytest-mock==3.12.0
+
+# Development
+black==23.9.1
+isort==5.12.0
+flake8==6.1.0
+sqlalchemy==2.0.23 
+
+# Bittensor
+bittensor==9.9.0
+
+rich
+yarl==1.22.0
+
+# Video hashing
+opencv-python-headless>=4.8.0
+numpy>=1.24.0

miner/upscaling/Dockerfile

@@ -0,0 +1,27 @@
+FROM python:3.11-slim
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    curl \
+    ca-certificates \
+    gnupg \
+    lsb-release \
+    && install -m 0755 -d /etc/apt/keyrings \
+    && curl -fsSL https://download.docker.com/linux/debian/gpg \
+       | gpg --dearmor -o /etc/apt/keyrings/docker.gpg \
+    && echo "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.gpg] \
+       https://download.docker.com/linux/debian $(lsb_release -cs) stable" \
+       > /etc/apt/sources.list.d/docker.list \
+    && apt-get update && apt-get install -y --no-install-recommends docker-ce-cli \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY app.py .
+
+RUN mkdir -p /tmp/organic-proxy
+
+EXPOSE 8003
+CMD ["python3", "-m", "uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8003"]

miner/upscaling/SETUP.md

@@ -0,0 +1,15 @@
+```sh
+# nvidia-container-toolkit
+curl -fsSL https://nvidia.github.io/libnvidia-container/gpgkey | sudo gpg --dearmor -o /usr/share/keyrings/nvidia-container-toolkit-keyring.gpg
+curl -s -L https://nvidia.github.io/libnvidia-container/stable/deb/nvidia-container-toolkit.list | \
+  sed 's#deb https://#deb [signed-by=/usr/share/keyrings/nvidia-container-toolkit-keyring.gpg] https://#g' | \
+  sudo tee /etc/apt/sources.list.d/nvidia-container-toolkit.list
+sudo apt update && sudo apt install -y nvidia-container-toolkit
+sudo nvidia-ctk runtime configure --runtime=docker
+sudo systemctl restart docker
+
+# Pull video2x image
+docker pull ghcr.io/k4yt3x/video2x:6.4.0
+```
+
+The Video2X worker runs with `--network none` and `--pull never`, so this image must be present locally before starting the `upscaling-video2x` profile.

miner/upscaling/app.py

@@ -0,0 +1,333 @@
+"""
+Upscaling microservice — wraps video2x (ghcr.io/k4yt3x/video2x:6.4.0).
+
+Accepts a video path (on shared volume) OR a URL and scale factor (2 or 4),
+runs video2x via Docker CLI, returns the output path or uploaded S3 URL.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import subprocess
+import uuid
+from typing import Optional
+
+import boto3
+import httpx
+from botocore.config import Config
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, Field
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s | %(levelname)-7s | %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+log = logging.getLogger("upscaling")
+
+app = FastAPI(title="Video Upscaling Service")
+
+VIDEO2X_IMAGE = os.getenv("VIDEO2X_IMAGE", "ghcr.io/k4yt3x/video2x:6.4.0")
+SHARED_VOLUME_PATH = os.getenv("SHARED_VOLUME_PATH", "/tmp/organic-proxy")
+DOCKER_VOLUME_NAME = os.getenv("DOCKER_VOLUME_NAME", "")
+HOST_SHARED_VOLUME_PATH = os.getenv("HOST_SHARED_VOLUME_PATH", "").strip()
+DISABLE_REMOTE_IO = os.getenv("DISABLE_REMOTE_IO", "false").lower() in ("1", "true", "yes")
+MAX_CONCURRENT = int(os.getenv("MAX_CONCURRENT_UPSCALING", "2"))
+
+# Storage provider label only. Uploads use one S3-compatible code path.
+STORAGE_PROVIDER = os.getenv("MINER_STORAGE_PROVIDER", "s3").lower()
+S3_REGION = os.getenv("MINER_STORAGE_S3_REGION", "us-east-1").strip() or "us-east-1"
+S3_BUCKET = os.getenv("MINER_STORAGE_S3_BUCKET_NAME", "").strip()
+S3_ACCESS_KEY_ID = os.getenv("MINER_STORAGE_S3_ACCESS_KEY_ID", "").strip()
+S3_SECRET_ACCESS_KEY = os.getenv("MINER_STORAGE_S3_SECRET_ACCESS_KEY", "").strip()
+S3_ENDPOINT_URL = (
+    os.getenv("MINER_STORAGE_S3_ENDPOINT_URL")
+    or os.getenv("MINER_STORAGE_S3_ENDPOINT")
+    or ""
+).strip()
+S3_PRESIGNED_EXPIRY = int(
+    os.getenv("MINER_STORAGE_S3_PRESIGNED_EXPIRY")
+    or os.getenv("S3_PRESIGNED_EXPIRY")
+    or "3600"
+)
+
+_semaphore = asyncio.Semaphore(MAX_CONCURRENT)
+_queue_size = 0
+_active_count = 0
+_lock = asyncio.Lock()
+
+
+def _is_url(path: str) -> bool:
+    return path.startswith("http://") or path.startswith("https://")
+
+
+def _get_s3_client():
+    client_kwargs = {
+        "region_name": S3_REGION,
+        "aws_access_key_id": S3_ACCESS_KEY_ID or None,
+        "aws_secret_access_key": S3_SECRET_ACCESS_KEY or None,
+        "config": Config(signature_version="s3v4"),
+    }
+    if S3_ENDPOINT_URL:
+        client_kwargs["endpoint_url"] = S3_ENDPOINT_URL
+    return boto3.client("s3", **client_kwargs)
+
+
+def _storage_config_status() -> dict[str, object]:
+    return {
+        "provider": STORAGE_PROVIDER,
+        "region": S3_REGION,
+        "bucket_configured": bool(S3_BUCKET),
+        "access_key_configured": bool(S3_ACCESS_KEY_ID),
+        "secret_key_configured": bool(S3_SECRET_ACCESS_KEY),
+        "endpoint_configured": bool(S3_ENDPOINT_URL),
+    }
+
+
+def _validate_s3_config():
+    missing = []
+    if not S3_BUCKET:
+        missing.append("MINER_STORAGE_S3_BUCKET_NAME")
+    if not S3_ACCESS_KEY_ID:
+        missing.append("MINER_STORAGE_S3_ACCESS_KEY_ID")
+    if not S3_SECRET_ACCESS_KEY:
+        missing.append("MINER_STORAGE_S3_SECRET_ACCESS_KEY")
+
+    if missing:
+        raise RuntimeError(f"Missing storage configuration: {', '.join(missing)}")
+
+
+async def _download_url(url: str, dest: str):
+    """Download a URL (presigned S3 or direct) to local path."""
+    log.info(f"Downloading {url[:80]}... → {dest}")
+    async with httpx.AsyncClient(timeout=600.0, follow_redirects=True) as client:
+        async with client.stream("GET", url) as resp:
+            resp.raise_for_status()
+            with open(dest, "wb") as f:
+                async for chunk in resp.aiter_bytes(chunk_size=8192):
+                    f.write(chunk)
+    log.info(f"Downloaded {os.path.getsize(dest) / (1024*1024):.1f} MB → {dest}")
+
+
+def _upload_to_s3(local_path: str, key: str) -> str:
+    """Upload file to storage and return presigned URL."""
+    _validate_s3_config()
+    client = _get_s3_client()
+    client.upload_file(local_path, S3_BUCKET, key)
+    url = client.generate_presigned_url(
+        "get_object",
+        Params={"Bucket": S3_BUCKET, "Key": key},
+        ExpiresIn=min(S3_PRESIGNED_EXPIRY, 604800),
+    )
+    log.info(f"Uploaded to {STORAGE_PROVIDER} storage: s3://{S3_BUCKET}/{key}")
+    return url
+
+
+class UpscaleRequest(BaseModel):
+    video_path: str = Field(..., description="Local path or URL to input video")
+    scale: int = Field(..., description="Upscaling factor: 2 or 4")
+    task_id: str = Field("", description="Task ID for logging")
+    model: str = Field("realesr-animevideov3", description="RealESRGAN model name")
+    codec: str = Field("hevc_nvenc", description="Output video codec")
+    cq: int = Field(35, description="Constant quality value")
+
+
+class UpscaleResponse(BaseModel):
+    output_path: str = Field("", description="Local path (local mode) or empty (remote mode)")
+    output_url: str = Field("", description="S3 presigned URL (remote mode) or empty (local mode)")
+    success: bool
+    error: Optional[str] = None
+    queue_position: Optional[int] = None
+    active_tasks: Optional[int] = None
+    queued_tasks: Optional[int] = None
+
+
+def _detect_volume_name() -> str:
+    if DOCKER_VOLUME_NAME:
+        return DOCKER_VOLUME_NAME
+    hostname = os.environ.get("HOSTNAME", "")
+    if hostname:
+        try:
+            result = subprocess.run(
+                ["docker", "inspect", "--format",
+                 '{{range .Mounts}}{{if eq .Destination "' + SHARED_VOLUME_PATH + '"}}{{.Name}}{{end}}{{end}}',
+                 hostname],
+                capture_output=True, text=True, timeout=10,
+            )
+            vol = result.stdout.strip()
+            if vol:
+                log.info(f"Auto-detected volume name: {vol}")
+                return vol
+        except Exception as e:
+            log.warning(f"Volume auto-detect failed: {e}")
+    fallback = "organic-proxy_video-tmp"
+    log.warning(f"Using fallback volume name: {fallback}")
+    return fallback
+
+
+_resolved_volume: Optional[str] = None
+
+
+def _get_volume_name() -> str:
+    global _resolved_volume
+    if _resolved_volume is None:
+        _resolved_volume = _detect_volume_name()
+    return _resolved_volume
+
+
+def _get_docker_mount_source() -> str:
+    if HOST_SHARED_VOLUME_PATH:
+        return HOST_SHARED_VOLUME_PATH
+    return _get_volume_name()
+
+
+@app.get("/health")
+async def health():
+    return {
+        "status": "ok",
+        "max_concurrent": MAX_CONCURRENT,
+        "active_tasks": _active_count,
+        "queued_tasks": max(0, _queue_size - _active_count),
+        "storage": _storage_config_status(),
+    }
+
+
+@app.get("/queue")
+async def queue_status():
+    return {
+        "max_concurrent": MAX_CONCURRENT,
+        "active_tasks": _active_count,
+        "queued_tasks": max(0, _queue_size - _active_count),
+        "total_pending": _queue_size,
+    }
+
+
+@app.post("/upscale", response_model=UpscaleResponse)
+async def upscale(req: UpscaleRequest):
+    if req.scale not in (2, 4):
+        raise HTTPException(status_code=400, detail="scale must be 2 or 4")
+
+    task_label = req.task_id or uuid.uuid4().hex[:8]
+    remote_mode = _is_url(req.video_path)
+
+    if remote_mode and DISABLE_REMOTE_IO:
+        return UpscaleResponse(success=False, error="Remote URL input is disabled for this local-only service")
+
+    # --- Resolve input to local path ---
+    if remote_mode:
+        local_input = os.path.join(SHARED_VOLUME_PATH, f"{task_label}_input.mp4")
+        try:
+            await _download_url(req.video_path, local_input)
+        except Exception as e:
+            return UpscaleResponse(success=False, error=f"Failed to download input: {e}")
+    else:
+        local_input = req.video_path
+        if not os.path.exists(local_input):
+            raise HTTPException(status_code=400, detail=f"Input file not found: {local_input}")
+
+    basename = os.path.splitext(os.path.basename(local_input))[0]
+    output_filename = f"{basename}_upscaled_{req.scale}x.mp4"
+    output_path = os.path.join(SHARED_VOLUME_PATH, output_filename)
+
+    # Build paths relative to container mount point
+    rel_input = os.path.relpath(local_input, SHARED_VOLUME_PATH)
+    rel_output = os.path.relpath(output_path, SHARED_VOLUME_PATH)
+
+    cmd = [
+        "docker", "run",
+        "--gpus", "all",
+        "-e", "NVIDIA_DRIVER_CAPABILITIES=video,compute,utility",
+        "--network", "none",
+        "--pull", "never",
+        "--rm",
+        "-v", f"{_get_docker_mount_source()}:/host",
+        VIDEO2X_IMAGE,
+        "-i", f"/host/{rel_input}",
+        "-o", f"/host/{rel_output}",
+        "-p", "realesrgan",
+        "-s", str(req.scale),
+        "--realesrgan-model", req.model,
+        "-c", req.codec,
+        "-e", "preset=p7",
+        "-e", f"cq={req.cq}",
+        "-e", "profile=main",
+        "-e", "pix_fmt=yuv420p",
+        "-e", "sar=1:1",
+        "-e", "color_primaries=bt709",
+        "-e", "color_trc=bt709",
+        "-e", "colorspace=bt709",
+        "-e", "movflags=+faststart",
+    ]
+
+    global _queue_size, _active_count
+
+    async with _lock:
+        _queue_size += 1
+        position = _queue_size
+        queued = max(0, _queue_size - _active_count)
+
+    log.info(f"[{task_label}] Queued {req.scale}x upscale (position={position}, waiting={queued}, remote={remote_mode})")
+
+    try:
+        async with _semaphore:
+            async with _lock:
+                _active_count += 1
+                queued_now = max(0, _queue_size - _active_count)
+
+            log.info(f"[{task_label}] Starting {req.scale}x upscale (active={_active_count}/{MAX_CONCURRENT}, queued={queued_now})")
+            log.info(f"[{task_label}] Command: {' '.join(cmd)}")
+
+            proc = await asyncio.create_subprocess_exec(
+                *cmd,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            stdout, stderr = await proc.communicate()
+
+            async with _lock:
+                _active_count -= 1
+    finally:
+        async with _lock:
+            _queue_size -= 1
+
+    stats = dict(active_tasks=_active_count, queued_tasks=max(0, _queue_size - _active_count))
+
+    if proc.returncode != 0:
+        err_msg = stderr.decode(errors="replace").strip()
+        log.error(f"[{task_label}] video2x failed (rc={proc.returncode}): {err_msg}")
+        if remote_mode:
+            _cleanup(local_input)
+        return UpscaleResponse(success=False, error=err_msg, **stats)
+
+    if not os.path.exists(output_path):
+        log.error(f"[{task_label}] Output file not found after video2x: {output_path}")
+        if remote_mode:
+            _cleanup(local_input)
+        return UpscaleResponse(success=False, error="Output file not created", **stats)
+
+    # --- Remote mode: upload result to S3, return URL ---
+    if remote_mode:
+        try:
+            s3_key = f"processing/{task_label}/{output_filename}"
+            output_url = _upload_to_s3(output_path, s3_key)
+            log.info(f"[{task_label}] Upscaling complete (remote): {output_url}")
+            return UpscaleResponse(output_url=output_url, success=True, **stats)
+        except Exception as e:
+            log.error(f"[{task_label}] S3 upload failed: {e}")
+            return UpscaleResponse(success=False, error=f"S3 upload failed: {e}", **stats)
+        finally:
+            _cleanup(local_input, output_path)
+
+    log.info(f"[{task_label}] Upscaling complete (local): {output_path}")
+    return UpscaleResponse(output_path=output_path, success=True, **stats)
+
+
+def _cleanup(*paths: str):
+    for p in paths:
+        if p and os.path.exists(p):
+            try:
+                os.remove(p)
+            except OSError:
+                pass

miner/upscaling/ffmpeg/Dockerfile

@@ -0,0 +1,67 @@
+FROM nvidia/cuda:13.0.0-devel-ubuntu22.04
+
+# Build dependencies for ffmpeg + NVENC/NVDEC + codec libraries
+RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
+    build-essential ninja-build nasm perl g++ cmake wget unzip \
+    nvidia-cuda-dev nvidia-cuda-toolkit pkg-config \
+    libx264-dev libx265-dev libvpx-dev \
+    git python3-pip python3-venv curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install meson (needed for dav1d build)
+RUN pip3 install --no-cache-dir meson
+
+# nv-codec-headers (NVENC/NVDEC API headers)
+RUN wget https://github.com/FFmpeg/nv-codec-headers/archive/876af32a202d0de83bd1d36fe74ee0f7fcf86b0d.zip \
+    && unzip 876af32a202d0de83bd1d36fe74ee0f7fcf86b0d.zip \
+    && cd nv-codec-headers-876af32a202d0de83bd1d36fe74ee0f7fcf86b0d \
+    && make && make install && cd .. \
+    && rm -rf nv-codec-headers-* 876af32a202d0de83bd1d36fe74ee0f7fcf86b0d.zip
+
+# dav1d (AV1 decoder)
+RUN git clone --depth 1 --branch 1.4.2 https://code.videolan.org/videolan/dav1d.git \
+    && cd dav1d && mkdir build && cd build \
+    && meson setup .. --default-library=static --buildtype release \
+    && ninja && ninja install && ldconfig \
+    && cd ../.. && rm -rf dav1d
+
+# SVT-AV1 (AV1 encoder for libsvtav1)
+RUN git clone --depth 1 --branch v2.3.0 https://gitlab.com/AOMediaCodec/SVT-AV1.git \
+    && cd SVT-AV1 && mkdir build && cd build \
+    && cmake .. -G"Unix Makefiles" -DCMAKE_BUILD_TYPE=Release -DBUILD_SHARED_LIBS=ON \
+    && make -j$(nproc) && make install && ldconfig \
+    && cd ../.. && rm -rf SVT-AV1
+
+# Build ffmpeg (pinned commit 33b215d) with CUDA + x264 + x265 + dav1d + svtav1
+RUN export PKG_CONFIG_PATH="/usr/local/lib/pkgconfig:/usr/local/lib/x86_64-linux-gnu/pkgconfig:$PKG_CONFIG_PATH" \
+    && git init ffmpeg && cd ffmpeg \
+    && git remote add origin https://github.com/FFmpeg/FFmpeg.git \
+    && git fetch --depth 1 origin 33b215d1554a14e87416a24f8e6034312e629af7 \
+    && git checkout FETCH_HEAD \
+    && ./configure \
+        --enable-gpl \
+        --enable-libx264 \
+        --enable-libx265 \
+        --enable-libvpx \
+        --enable-libdav1d \
+        --enable-libsvtav1 \
+        --enable-nonfree \
+        --enable-cuda-nvcc \
+        --enable-nvenc \
+        --enable-nvdec \
+        --extra-cflags="-I/usr/local/cuda/include" \
+        --extra-ldflags="-L/usr/local/cuda/lib64" \
+        --disable-static \
+        --enable-shared \
+        --nvccflags="-gencode arch=compute_75,code=sm_75 -gencode arch=compute_75,code=compute_75 -O2" \
+    && make -j$(nproc) && make install && ldconfig \
+    && cd .. && rm -rf ffmpeg
+
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+COPY app.py .
+RUN mkdir -p /tmp/organic-proxy
+
+EXPOSE 8003
+CMD ["python3", "-m", "uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8003"]

miner/upscaling/ffmpeg/SETUP.md

@@ -0,0 +1,12 @@
+# FFmpeg Upscaling Setup
+
+This variant uses FFmpeg directly (CUDA/NVENC capable build) for video upscaling.
+
+## Notes
+- The Dockerfile pins `nv-codec-headers`, `dav1d`, and FFmpeg commit `33b215d1554a14e87416a24f8e6034312e629af7`.
+- Requires NVIDIA GPU runtime on the host.
+
+## Run
+```bash
+docker compose --profile upscaling-ffmpeg up -d upscaling-ffmpeg
+```

miner/upscaling/ffmpeg/app.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import subprocess
+import uuid
+from typing import Optional
+
+import boto3
+import httpx
+from botocore.config import Config
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, Field
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)-7s | %(message)s")
+log = logging.getLogger("upscaling-ffmpeg")
+
+app = FastAPI(title="FFmpeg Upscaling Service")
+
+SHARED_VOLUME_PATH = os.getenv("SHARED_VOLUME_PATH", "/tmp/organic-proxy")
+MAX_CONCURRENT = int(os.getenv("MAX_CONCURRENT_UPSCALING", "2"))
+FFMPEG_CODEC = os.getenv("FFMPEG_CODEC", "hevc_nvenc")
+FFMPEG_PRESET = os.getenv("FFMPEG_PRESET", "p7")
+FFMPEG_CQ = int(os.getenv("FFMPEG_CQ", "35"))
+DISABLE_REMOTE_IO = os.getenv("DISABLE_REMOTE_IO", "false").lower() in ("1", "true", "yes")
+
+STORAGE_PROVIDER = os.getenv("MINER_STORAGE_PROVIDER", "s3").lower()
+S3_REGION = os.getenv("MINER_STORAGE_S3_REGION", "us-east-1").strip() or "us-east-1"
+S3_BUCKET = os.getenv("MINER_STORAGE_S3_BUCKET_NAME", "").strip()
+S3_ACCESS_KEY_ID = os.getenv("MINER_STORAGE_S3_ACCESS_KEY_ID", "").strip()
+S3_SECRET_ACCESS_KEY = os.getenv("MINER_STORAGE_S3_SECRET_ACCESS_KEY", "").strip()
+S3_ENDPOINT_URL = (os.getenv("MINER_STORAGE_S3_ENDPOINT_URL") or os.getenv("MINER_STORAGE_S3_ENDPOINT") or "").strip()
+S3_PRESIGNED_EXPIRY = int(os.getenv("MINER_STORAGE_S3_PRESIGNED_EXPIRY") or os.getenv("S3_PRESIGNED_EXPIRY") or "3600")
+
+_semaphore = asyncio.Semaphore(MAX_CONCURRENT)
+_queue_size = 0
+_active_count = 0
+_lock = asyncio.Lock()
+
+
+def _is_url(path: str) -> bool:
+    return path.startswith("http://") or path.startswith("https://")
+
+
+def _get_s3_client():
+    kwargs = {
+        "region_name": S3_REGION,
+        "aws_access_key_id": S3_ACCESS_KEY_ID or None,
+        "aws_secret_access_key": S3_SECRET_ACCESS_KEY or None,
+        "config": Config(signature_version="s3v4"),
+    }
+    if S3_ENDPOINT_URL:
+        kwargs["endpoint_url"] = S3_ENDPOINT_URL
+    return boto3.client("s3", **kwargs)
+
+
+def _validate_s3_config():
+    missing = [k for k, v in {
+        "MINER_STORAGE_S3_BUCKET_NAME": S3_BUCKET,
+        "MINER_STORAGE_S3_ACCESS_KEY_ID": S3_ACCESS_KEY_ID,
+        "MINER_STORAGE_S3_SECRET_ACCESS_KEY": S3_SECRET_ACCESS_KEY,
+    }.items() if not v]
+    if missing:
+        raise RuntimeError(f"Missing storage configuration: {', '.join(missing)}")
+
+
+async def _download_url(url: str, dest: str):
+    async with httpx.AsyncClient(timeout=600.0, follow_redirects=True) as client:
+        async with client.stream("GET", url) as resp:
+            resp.raise_for_status()
+            with open(dest, "wb") as f:
+                async for chunk in resp.aiter_bytes(chunk_size=8192):
+                    f.write(chunk)
+
+
+def _upload_to_s3(local_path: str, key: str) -> str:
+    _validate_s3_config()
+    client = _get_s3_client()
+    client.upload_file(local_path, S3_BUCKET, key)
+    return client.generate_presigned_url("get_object", Params={"Bucket": S3_BUCKET, "Key": key}, ExpiresIn=min(S3_PRESIGNED_EXPIRY, 604800))
+
+
+class UpscaleRequest(BaseModel):
+    video_path: str
+    scale: int = Field(..., description="Upscaling factor: 2 or 4")
+    task_id: str = ""
+
+
+class UpscaleResponse(BaseModel):
+    output_path: str = ""
+    output_url: str = ""
+    success: bool
+    error: Optional[str] = None
+
+
+@app.get("/health")
+async def health():
+    return {"status": "ok", "max_concurrent": MAX_CONCURRENT, "active_tasks": _active_count, "queued_tasks": max(0, _queue_size - _active_count)}
+
+
+@app.post("/upscale", response_model=UpscaleResponse)
+async def upscale(req: UpscaleRequest):
+    global _queue_size, _active_count
+
+    if req.scale not in (2, 4):
+        raise HTTPException(status_code=400, detail="scale must be 2 or 4")
+
+    task = req.task_id or uuid.uuid4().hex[:8]
+    remote = _is_url(req.video_path)
+    if remote and DISABLE_REMOTE_IO:
+        return UpscaleResponse(success=False, error="Remote URL input is disabled for this local-only service")
+
+    local_input = os.path.join(SHARED_VOLUME_PATH, f"{task}_input.mp4") if remote else req.video_path
+    if remote:
+        await _download_url(req.video_path, local_input)
+    elif not os.path.exists(local_input):
+        raise HTTPException(status_code=400, detail=f"Input file not found: {local_input}")
+
+    base = os.path.splitext(os.path.basename(local_input))[0]
+    output_path = os.path.join(SHARED_VOLUME_PATH, f"{base}_upscaled_{req.scale}x.mp4")
+
+    cmd = [
+        "ffmpeg", "-y", "-hwaccel", "cuda", "-i", local_input,
+        "-vf", f"scale=iw*{req.scale}:ih*{req.scale}:flags=lanczos",
+        "-c:v", FFMPEG_CODEC, "-preset", FFMPEG_PRESET, "-cq", str(FFMPEG_CQ),
+        "-c:a", "copy", output_path,
+    ]
+
+    async with _lock:
+        _queue_size += 1
+    async with _semaphore:
+        async with _lock:
+            _active_count += 1
+        try:
+            result = await asyncio.to_thread(subprocess.run, cmd, capture_output=True, text=True)
+            if result.returncode != 0:
+                return UpscaleResponse(success=False, error=result.stderr[-1000:])
+            if remote:
+                key = f"upscaling/{task}/{os.path.basename(output_path)}"
+                url = _upload_to_s3(output_path, key)
+                return UpscaleResponse(success=True, output_url=url)
+            return UpscaleResponse(success=True, output_path=output_path)
+        finally:
+            async with _lock:
+                _active_count = max(0, _active_count - 1)
+                _queue_size = max(0, _queue_size - 1)

miner/upscaling/ffmpeg/requirements.txt

@@ -0,0 +1,5 @@
+fastapi==0.111.0
+uvicorn==0.30.1
+boto3==1.34.122
+httpx==0.27.0
+pydantic==2.8.2

miner/upscaling/requirements.txt

@@ -0,0 +1,5 @@
+fastapi==0.115.12
+uvicorn==0.34.2
+pydantic==2.11.1
+boto3==1.35.99
+httpx==0.28.1

neurons/miner.py

@@ -1,11 +1,18 @@
+import asyncio
+import os
+import posixpath
 import time
-import uuid
 import traceback
-import os
-import asyncio
+import uuid
+from pathlib import Path
 from typing import Tuple
-from loguru import logger
+
+import boto3
+import httpx
 import bittensor as bt
+from botocore.config import Config as BotoConfig
+from dotenv import load_dotenv
+from loguru import logger
 from vidaio_subnet_core.base.miner import BaseMiner
 from vidaio_subnet_core.protocol import (
     VideoUpscalingProtocol,
@@ -21,13 +28,58 @@ from vidaio_subnet_core.protocol import (
     JobKickoffResponse,
     PollResponse,
 )
-from services.miner_utilities.miner_utils import video_upscaler, video_compressor
 
 from vidaio_subnet_core.utilities.version import check_version
 
+REPO_ROOT = Path(__file__).resolve().parents[1]
+load_dotenv(REPO_ROOT / "miner" / ".env", override=False)
+
 MAX_CONTENT_LEN = ContentLength.FIVE
 warrant_task = TaskType.UPSCALING
 
+UPSCALING_SERVICE_URL = os.getenv("MINER_UPSCALING_SERVICE_URL", "http://localhost:8003").rstrip("/")
+COMPRESSION_SERVICE_URL = os.getenv("MINER_COMPRESSION_SERVICE_URL", "http://localhost:8004").rstrip("/")
+UPSCALING_SERVICE_TIMEOUT_SECONDS = float(os.getenv("MINER_UPSCALING_SERVICE_TIMEOUT_SECONDS", "1200"))
+COMPRESSION_SERVICE_TIMEOUT_SECONDS = float(os.getenv("MINER_COMPRESSION_SERVICE_TIMEOUT_SECONDS", "600"))
+MINER_DOWNLOAD_TIMEOUT_SECONDS = float(os.getenv("MINER_DOWNLOAD_TIMEOUT_SECONDS", "600"))
+MINER_UPLOAD_URL_EXPIRY_SECONDS = int(os.getenv("MINER_STORAGE_S3_PRESIGNED_EXPIRY", os.getenv("S3_PRESIGNED_EXPIRY", "3600")))
+
+HOST_SHARED_VOLUME_PATH = Path(
+    os.getenv("MINER_SHARED_VOLUME_PATH")
+    or os.getenv("MINER_SHARED_DIR", "/tmp/vidaio-miner-video-tmp")
+).expanduser()
+if not HOST_SHARED_VOLUME_PATH.is_absolute():
+    HOST_SHARED_VOLUME_PATH = REPO_ROOT / HOST_SHARED_VOLUME_PATH
+CONTAINER_SHARED_VOLUME_PATH = os.getenv("MINER_CONTAINER_SHARED_VOLUME_PATH", "/tmp/organic-proxy").rstrip("/")
+
+S3_REGION = os.getenv("MINER_STORAGE_S3_REGION", "us-east-1").strip() or "us-east-1"
+S3_BUCKET = os.getenv("MINER_STORAGE_S3_BUCKET_NAME", "").strip()
+S3_ACCESS_KEY_ID = os.getenv("MINER_STORAGE_S3_ACCESS_KEY_ID", "").strip()
+S3_SECRET_ACCESS_KEY = os.getenv("MINER_STORAGE_S3_SECRET_ACCESS_KEY", "").strip()
+S3_ENDPOINT_URL = (
+    os.getenv("MINER_STORAGE_S3_ENDPOINT_URL")
+    or os.getenv("MINER_STORAGE_S3_ENDPOINT")
+    or ""
+).strip()
+
+TASK_TYPE_TO_SCALE = {
+    "SD24K": 4,
+    "HD24K": 2,
+    "SD2HD": 2,
+    "4K28K": 2,
+}
+
+COMPRESSION_CQ_BY_TYPE = {
+    "Low": 40,
+    "Medium": 35,
+    "High": 30,
+}
+
+
+def _is_url(path: str) -> bool:
+    return path.startswith("http://") or path.startswith("https://")
+
+
 class Miner(BaseMiner):
     def __init__(self, config: dict | None = None) -> None:
         """
@@ -38,6 +90,219 @@ class Miner(BaseMiner):
         # Structure: { job_id: {"task": asyncio.Task, "result": str | None, "error": str | None} }
         self._jobs: dict[str, dict] = {}
 
+    def _container_shared_path(self, host_path: Path) -> str:
+        return f"{CONTAINER_SHARED_VOLUME_PATH}/{host_path.name}"
+
+    def _host_shared_path(self, container_path: str) -> Path:
+        shared_mount = posixpath.normpath(CONTAINER_SHARED_VOLUME_PATH)
+        normalized_path = posixpath.normpath(container_path)
+
+        if normalized_path == shared_mount or not normalized_path.startswith(f"{shared_mount}/"):
+            raise ValueError(f"Service returned path outside shared volume: {container_path}")
+
+        relative_path = posixpath.relpath(normalized_path, shared_mount)
+        host_path = (HOST_SHARED_VOLUME_PATH / relative_path).resolve()
+        shared_root = HOST_SHARED_VOLUME_PATH.resolve()
+
+        try:
+            host_path.relative_to(shared_root)
+        except ValueError:
+            raise ValueError(f"Mapped host path escapes shared volume: {host_path}") from None
+
+        return host_path
+
+    async def _download_to_shared_volume(self, video_url: str, task_id: str) -> tuple[Path, str]:
+        HOST_SHARED_VOLUME_PATH.mkdir(parents=True, exist_ok=True)
+        host_path = HOST_SHARED_VOLUME_PATH / f"{task_id}_input.mp4"
+
+        logger.info(f"Downloading validator payload to shared volume: {host_path}")
+        timeout = httpx.Timeout(MINER_DOWNLOAD_TIMEOUT_SECONDS)
+        try:
+            async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
+                async with client.stream("GET", video_url) as response:
+                    response.raise_for_status()
+                    with open(host_path, "wb") as file:
+                        async for chunk in response.aiter_bytes(chunk_size=1024 * 1024):
+                            file.write(chunk)
+        except Exception:
+            self._cleanup_shared_files(host_path)
+            raise
+
+        return host_path, self._container_shared_path(host_path)
+
+    def _get_s3_client(self):
+        client_kwargs = {
+            "region_name": S3_REGION,
+            "aws_access_key_id": S3_ACCESS_KEY_ID or None,
+            "aws_secret_access_key": S3_SECRET_ACCESS_KEY or None,
+            "config": BotoConfig(signature_version="s3v4"),
+        }
+        if S3_ENDPOINT_URL:
+            client_kwargs["endpoint_url"] = S3_ENDPOINT_URL
+        return boto3.client("s3", **client_kwargs)
+
+    async def _upload_processed_video(self, host_path: Path, service_name: str, task_id: str) -> str | None:
+        missing = []
+        if not S3_BUCKET:
+            missing.append("MINER_STORAGE_S3_BUCKET_NAME")
+        if not S3_ACCESS_KEY_ID:
+            missing.append("MINER_STORAGE_S3_ACCESS_KEY_ID")
+        if not S3_SECRET_ACCESS_KEY:
+            missing.append("MINER_STORAGE_S3_SECRET_ACCESS_KEY")
+
+        if missing:
+            logger.error(f"Missing storage configuration: {', '.join(missing)}")
+            return None
+
+        if not host_path.exists():
+            logger.error(f"Processed file not found on shared volume: {host_path}")
+            return None
+
+        object_key = f"processing/{service_name}/{task_id}/{host_path.name}"
+
+        def _upload_and_sign() -> str:
+            client = self._get_s3_client()
+            client.upload_file(str(host_path), S3_BUCKET, object_key)
+            return client.generate_presigned_url(
+                "get_object",
+                Params={"Bucket": S3_BUCKET, "Key": object_key},
+                ExpiresIn=min(MINER_UPLOAD_URL_EXPIRY_SECONDS, 604800),
+            )
+
+        logger.info(f"Uploading processed {service_name} output: {object_key}")
+        return await asyncio.to_thread(_upload_and_sign)
+
+    def _cleanup_shared_files(self, *paths: Path | None):
+        for path in paths:
+            if path and path.exists():
+                try:
+                    path.unlink()
+                except OSError as e:
+                    logger.warning(f"Failed to remove shared file {path}: {e}")
+
+    def _upscaling_service_payload(self, video_path: str, task_type: str, task_id: str) -> dict:
+        scale = TASK_TYPE_TO_SCALE.get(task_type, 2)
+        if task_type not in TASK_TYPE_TO_SCALE:
+            logger.warning(f"Unknown upscaling task_type={task_type}, defaulting to 2x scale")
+
+        return {
+            "video_path": video_path,
+            "scale": scale,
+            "task_id": task_id,
+        }
+
+    def _compression_type_from_payload(self, payload) -> str:
+        task_data = payload.model_dump() if hasattr(payload, "model_dump") else {}
+        compression_type = task_data.get("compression_type") or getattr(payload, "compression_type", None)
+        if compression_type in COMPRESSION_CQ_BY_TYPE:
+            return compression_type
+
+        vmaf_threshold = float(payload.vmaf_threshold)
+        if vmaf_threshold >= 93:
+            return "High"
+        if vmaf_threshold >= 89:
+            return "Medium"
+        return "Low"
+
+    def _compression_service_payload(self, payload, video_path: str, task_id: str) -> dict:
+        compression_type = self._compression_type_from_payload(payload)
+        compression_config = {
+            "video_path": video_path,
+            "task_id": task_id,
+            "codec": payload.target_codec,
+            "codec_mode": payload.codec_mode.upper(),
+            "target_bitrate": int(float(payload.target_bitrate) * 1_000_000),
+        }
+
+        if compression_type == "Low":
+            compression_config["cq"] = 40
+        elif compression_type == "Medium":
+            compression_config["cq"] = 35
+        elif compression_type == "High":
+            compression_config["cq"] = 30
+
+        return compression_config
+
+    async def _post_processing_service(
+        self,
+        service_name: str,
+        service_url: str,
+        endpoint: str,
+        payload: dict,
+        timeout_seconds: float,
+    ) -> str | None:
+        url = f"{service_url}{endpoint}"
+        timeout = httpx.Timeout(timeout_seconds)
+
+        logger.info(f"Forwarding {service_name} request to {url}")
+        async with httpx.AsyncClient(timeout=timeout) as client:
+            response = await client.post(url, json=payload)
+
+        if response.status_code != 200:
+            logger.error(f"{service_name} service error {response.status_code}: {response.text[:1000]}")
+            return None
+
+        result = response.json()
+        if not result.get("success", False):
+            logger.error(f"{service_name} service failed: {result.get('error')}")
+            return None
+
+        processed_video_url = result.get("output_url") or result.get("output_path")
+        if not processed_video_url:
+            logger.error(f"{service_name} service returned no output URL/path")
+            return None
+
+        logger.info(f"Received {service_name} result from container service")
+        return processed_video_url
+
+    async def _forward_upscaling_to_service(self, payload_url: str, task_type: str) -> str | None:
+        task_id = uuid.uuid4().hex[:12]
+        input_host_path = None
+        output_host_path = None
+        try:
+            input_host_path, input_container_path = await self._download_to_shared_volume(payload_url, task_id)
+            processed_ref = await self._post_processing_service(
+                "upscaling",
+                UPSCALING_SERVICE_URL,
+                "/upscale",
+                self._upscaling_service_payload(input_container_path, task_type, task_id),
+                UPSCALING_SERVICE_TIMEOUT_SECONDS,
+            )
+            if processed_ref is None:
+                return None
+            if _is_url(processed_ref):
+                return processed_ref
+
+            output_host_path = self._host_shared_path(processed_ref)
+            return await self._upload_processed_video(output_host_path, "upscaling", task_id)
+        finally:
+            self._cleanup_shared_files(input_host_path, output_host_path)
+
+    async def _forward_compression_to_service(self, payload) -> str | None:
+        task_id = uuid.uuid4().hex[:12]
+        input_host_path = None
+        output_host_path = None
+        try:
+            input_host_path, input_container_path = await self._download_to_shared_volume(
+                payload.reference_video_url, task_id
+            )
+            processed_ref = await self._post_processing_service(
+                "compression",
+                COMPRESSION_SERVICE_URL,
+                "/compress",
+                self._compression_service_payload(payload, input_container_path, task_id),
+                COMPRESSION_SERVICE_TIMEOUT_SECONDS,
+            )
+            if processed_ref is None:
+                return None
+            if _is_url(processed_ref):
+                return processed_ref
+
+            output_host_path = self._host_shared_path(processed_ref)
+            return await self._upload_processed_video(output_host_path, "compression", task_id)
+        finally:
+            self._cleanup_shared_files(input_host_path, output_host_path)
+
     async def forward_upscaling_requests(self, synapse: VideoUpscalingProtocol) -> VideoUpscalingProtocol:
         """
         Processes a video upscaling request by downloading, upscaling,
@@ -55,7 +320,7 @@ class Miner(BaseMiner):
         check_version(synapse.version)
 
         try:
-            processed_video_url = await video_upscaler(payload_url, task_type)
+            processed_video_url = await self._forward_upscaling_to_service(payload_url, task_type)
             
             if processed_video_url is None:
                 logger.info(f"💔 Failed to upscaling video 💔")
@@ -82,7 +347,6 @@ class Miner(BaseMiner):
 
         start_time = time.time()
 
-        payload_url: str = synapse.miner_payload.reference_video_url
         vmaf_threshold: float = synapse.miner_payload.vmaf_threshold
         target_codec: str = synapse.miner_payload.target_codec
         codec_mode: str = synapse.miner_payload.codec_mode
@@ -94,7 +358,7 @@ class Miner(BaseMiner):
         check_version(synapse.version)
 
         try:
-            processed_video_url = await video_compressor(payload_url, vmaf_threshold, target_codec, codec_mode, target_bitrate)
+            processed_video_url = await self._forward_compression_to_service(synapse.miner_payload)
 
             if processed_video_url is None:
                 logger.info(f"💔 Failed to compress video 💔")
@@ -159,16 +423,10 @@ class Miner(BaseMiner):
                 synapse.job_response = JobKickoffResponse(accepted=False)
                 return synapse
 
-            payload_url = synapse.miner_payload.reference_video_url
-            vmaf_threshold = synapse.miner_payload.vmaf_threshold
-            target_codec = synapse.miner_payload.target_codec
-            codec_mode = synapse.miner_payload.codec_mode
-            target_bitrate = synapse.miner_payload.target_bitrate
-
             async def _run_compression():
-                return await video_compressor(
-                    payload_url, vmaf_threshold, target_codec, codec_mode, target_bitrate
-                )
+                result = await self._forward_compression_to_service(synapse.miner_payload)
+                logger.info(f"CompressionJob processed successfully | job_id={job_id} | result={result}")
+                return result
 
             task = asyncio.create_task(_run_compression())
             self._jobs[job_id] = {"task": task, "result": None, "error": None}
@@ -275,7 +533,9 @@ class Miner(BaseMiner):
             task_type = synapse.miner_payload.task_type
 
             async def _run_upscaling():
-                return await video_upscaler(payload_url, task_type)
+                result = await self._forward_upscaling_to_service(payload_url, task_type)
+                logger.info(f"UpscalingJob processed successfully | job_id={job_id} | result={result}")
+                return result
 
             task = asyncio.create_task(_run_upscaling())
             self._jobs[job_id] = {"task": task, "result": None, "error": None}