Creepso/fluxa-core
1
0
Fork 0
mirror of https://github.com/FluxaMedia/fluxa-core.git synced 2026-08-09 00:17:34 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

chore: consolidate production hardening updates

Browse source
This commit is contained in:
KhooLy 2026-08-01 18:52:22 +03:00
parent 6b885bd078
commit d33fd06fc9
27 changed files with 735 additions and 232 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

49
.github/workflows/ci.yml vendored
View file

@ -4,6 +4,8 @@ on:
push:
branches: [master]
pull_request:
schedule:
- cron: '17 3 * * 1'
env:
CARGO_TERM_COLOR: always
@ -15,7 +17,8 @@ jobs:
- uses: actions/checkout@v7
- uses: dtolnay/rust-toolchain@stable
- uses: Swatinem/rust-cache@v2
- run: cargo test --workspace --lib
- run: cargo test --workspace --all-targets
- run: cargo test --doc
clippy:
runs-on: ubuntu-latest
@ -25,7 +28,7 @@ jobs:
with:
components: clippy
- uses: Swatinem/rust-cache@v2
- run: cargo clippy --workspace --lib
- run: cargo clippy --workspace --all-targets -- -D warnings
bench-check:
runs-on: ubuntu-latest
@ -43,6 +46,48 @@ jobs:
- uses: Swatinem/rust-cache@v2
- run: cargo check --no-default-features --features wasm
feature-check:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
features: [streaming-shared, desktop, uniffi-bindings]
steps:
- uses: actions/checkout@v7
- uses: dtolnay/rust-toolchain@stable
- uses: Swatinem/rust-cache@v2
- run: cargo check --no-default-features --features ${{ matrix.features }}
android-check:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v7
- uses: dtolnay/rust-toolchain@stable
with:
targets: aarch64-linux-android,armv7-linux-androideabi,x86_64-linux-android
- uses: Swatinem/rust-cache@v2
- run: cargo check --target aarch64-linux-android
- run: cargo check --target armv7-linux-androideabi
- run: cargo check --target x86_64-linux-android
dependency-audit:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v7
- uses: EmbarkStudios/cargo-deny-action@v2
fuzz-smoke:
if: github.event_name == 'schedule'
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v7
- uses: dtolnay/rust-toolchain@stable
- run: cargo install cargo-fuzz --locked
- run: cd fuzz && cargo fuzz run episode_matching -- -runs=1000
- run: cd fuzz && cargo fuzz run percent_decode -- -runs=1000
- run: cd fuzz && cargo fuzz run manifest_parse -- -runs=1000
- run: cd fuzz && cargo fuzz run engine_dispatch -- -runs=1000
fmt:
runs-on: ubuntu-latest
steps:

5
fluxa-streaming-engine/src/dv_rewrite.rs
View file

@ -188,7 +188,10 @@ pub(crate) fn start_dv_rewrite_local_stream_server(
};
let thread = thread::spawn(move || {
let Ok(runtime) = tokio::runtime::Builder::new_current_thread().enable_all().build() else {
let Ok(runtime) = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
else {
return;
};
runtime.block_on(async move {

5
fluxa-streaming-engine/src/local_stream.rs
View file

@ -245,7 +245,10 @@ pub(crate) fn start_local_stream_server(
port,
};
let thread = thread::spawn(move || {
let Ok(runtime) = tokio::runtime::Builder::new_current_thread().enable_all().build() else {
let Ok(runtime) = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
else {
return;
};
runtime.block_on(async move {

149
fluxa-streaming-engine/src/torrent_engine.rs
View file

@ -13,14 +13,14 @@ use librqbit::{
use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::{HashMap, HashSet};
use std::io::SeekFrom;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::io::SeekFrom;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::task::{Context, Poll};
use std::thread;
use std::time::{Duration, Instant};
use tokio::io::{AsyncReadExt, AsyncSeekExt};
@ -95,6 +95,21 @@ struct PlaybackSession {
cancel: CancellationToken,
}
#[derive(Default, Clone, Copy)]
struct PlaybackTelemetry {
first_frame_ms: Option<u64>,
stall_count: u64,
stall_duration_ms: u64,
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct TelemetryEvent {
link: String,
event: String,
elapsed_ms: Option<u64>,
}
struct CancellableReader<R> {
inner: R,
// Keep the cancellation future alive while the underlying rqbit reader is
@ -159,6 +174,7 @@ struct EngineState {
prioritized_files: Arc<Mutex<HashMap<usize, TorrentFileFocus>>>,
playback_windows: Arc<Mutex<HashMap<(usize, usize), PlaybackWindow>>>,
playback_sessions: Arc<Mutex<HashMap<(usize, usize), PlaybackSession>>>,
playback_telemetry: Arc<Mutex<HashMap<usize, PlaybackTelemetry>>>,
/// Root cancellation per torrent. Probe readers use child tokens too, so
/// they cannot outlive playback deactivation or cache eviction.
torrent_cancellations: Arc<Mutex<HashMap<usize, CancellationToken>>>,
@ -322,6 +338,7 @@ pub fn start_torrent_server(
prioritized_files: Arc::new(Mutex::new(HashMap::new())),
playback_windows: Arc::new(Mutex::new(HashMap::new())),
playback_sessions: Arc::new(Mutex::new(HashMap::new())),
playback_telemetry: Arc::new(Mutex::new(HashMap::new())),
torrent_cancellations: Arc::new(Mutex::new(HashMap::new())),
lifecycle: Arc::new(Mutex::new(HashMap::new())),
active_torrent: Arc::new(Mutex::new(None)),
@ -337,6 +354,7 @@ pub fn start_torrent_server(
.route("/health", get(health))
.route("/settings", post(update_settings))
.route("/torrents", post(torrents))
.route("/telemetry", post(record_telemetry))
.route("/stream/fname", get(stream_fname))
.with_state(state);
@ -611,6 +629,37 @@ async fn torrents(
}
}
async fn record_telemetry(
State(state): State<EngineState>,
ConnectInfo(remote_addr): ConnectInfo<SocketAddr>,
Json(event): Json<TelemetryEvent>,
) -> Response {
if !request_authorized(&state, remote_addr, None) {
return error_response(StatusCode::UNAUTHORIZED, "unauthorized");
}
let Some(id) = lookup_known_link(&state, Some(&event.link)) else {
return error_response(StatusCode::NOT_FOUND, "torrent not found");
};
let mut telemetry = match state.playback_telemetry.lock() {
Ok(telemetry) => telemetry,
Err(_) => {
return error_response(StatusCode::INTERNAL_SERVER_ERROR, "telemetry unavailable");
}
};
let entry = telemetry.entry(id).or_default();
match event.event.as_str() {
"firstFrame" => entry.first_frame_ms = event.elapsed_ms.or(entry.first_frame_ms),
"stallStarted" => entry.stall_count = entry.stall_count.saturating_add(1),
"stallEnded" => {
entry.stall_duration_ms = entry
.stall_duration_ms
.saturating_add(event.elapsed_ms.unwrap_or_default())
}
_ => return error_response(StatusCode::BAD_REQUEST, "unsupported telemetry event"),
}
(StatusCode::OK, Json(json!({}))).into_response()
}
async fn stream_fname(
State(state): State<EngineState>,
Query(query): Query<StreamQuery>,
@ -954,10 +1003,7 @@ async fn status_response(
.and_then(|stats| stats.live.as_ref())
.map(|live| live.snapshot.peer_stats.seen)
.unwrap_or(0);
let peer_quality = state
.api
.api_peer_quality(TorrentIdOrHash::Id(id))
.ok();
let peer_quality = state.api.api_peer_quality(TorrentIdOrHash::Id(id)).ok();
let preload_size = state.preload_size.lock().map(|value| *value).unwrap_or(0);
let mut window = focus_file.and_then(|file_id| playback_window_for(state, id, file_id));
let playback_offset = window.map(|window| window.playback_offset).unwrap_or(0);
@ -972,7 +1018,9 @@ async fn status_response(
TorrentIdOrHash::Id(id),
file_id,
playback_offset,
window.map(|window| window.warm_ahead_bytes).unwrap_or(preload_size),
window
.map(|window| window.warm_ahead_bytes)
.unwrap_or(preload_size),
)
.ok()
.map(|response| response.contiguous_bytes)
@ -1002,8 +1050,29 @@ async fn status_response(
})
.unwrap_or(0.0);
let speed_to_bitrate_ratio = window
.map(|window| window.smoothed_download_bps * 8.0 / window.estimated_bitrate_bps.max(1) as f64)
.map(|window| {
window.smoothed_download_bps * 8.0 / window.estimated_bitrate_bps.max(1) as f64
})
.unwrap_or(0.0);
let playback_telemetry = state
.playback_telemetry
.lock()
.ok()
.and_then(|telemetry| telemetry.get(&id).copied())
.unwrap_or_default();
let scheduler = window.map(|window| {
json!({
"torrentId": window.torrent_id,
"fileId": window.file_id,
"urgentAheadBytes": window.urgent_ahead_bytes,
"warmAheadBytes": window.warm_ahead_bytes,
"contiguousReadyBytes": window.contiguous_ready_bytes,
"seekGeneration": window.seek_generation,
"seekElapsedMs": window.seek_started_at.map(|started| started.elapsed().as_millis() as u64),
"windowAgeMs": window.updated_at.elapsed().as_millis() as u64,
"wasReady": window.was_ready,
})
});
let stat = match stats.as_ref().map(|stats| stats.state) {
Some(TorrentStatsState::Live)
if loaded_size >= target_buffer_bytes && target_buffer_bytes > 0 =>
@ -1044,6 +1113,17 @@ async fn status_response(
"phase": playback_phase(stats.as_ref(), loaded_size, target_buffer_bytes, window),
"playback_offset": playback_offset,
"requested_end": window.map(|window| window.requested_end).unwrap_or(0),
"telemetry": {
"downloadSpeedBps": download_speed,
"speedToBitrateRatio": speed_to_bitrate_ratio,
"bufferedAheadBytes": loaded_size,
"bufferedAheadSeconds": buffered_ahead_seconds,
"phase": playback_phase(stats.as_ref(), loaded_size, target_buffer_bytes, window),
"scheduler": scheduler,
"firstFrameMs": playback_telemetry.first_frame_ms,
"stallCount": playback_telemetry.stall_count,
"stallDurationMs": playback_telemetry.stall_duration_ms
},
"file_stats": file_stats
}))
}
@ -1229,7 +1309,9 @@ async fn prewarm_reaper(state: EngineState) {
.api
.api_torrent_action_pause(TorrentIdOrHash::Id(torrent_id))
.await;
debug_log(format!("[TorrServer] paused idle prewarm torrent={torrent_id}"));
debug_log(format!(
"[TorrServer] paused idle prewarm torrent={torrent_id}"
));
}
enforce_cache_limit(&state).await;
}
@ -1239,7 +1321,9 @@ async fn enforce_cache_limit(state: &EngineState) {
let Some(limit) = state.cache_limit_bytes.lock().ok().and_then(|limit| *limit) else {
return;
};
let snapshots = state.api.api_torrent_list_ext(ApiTorrentListOpts { with_stats: true });
let snapshots = state
.api
.api_torrent_list_ext(ApiTorrentListOpts { with_stats: true });
let mut entries = state
.lifecycle
.lock()
@ -1254,7 +1338,11 @@ async fn enforce_cache_limit(state: &EngineState) {
id,
lifecycle.active,
lifecycle.last_accessed,
torrent.stats.as_ref().map(|stats| stats.progress_bytes).unwrap_or(0),
torrent
.stats
.as_ref()
.map(|stats| stats.progress_bytes)
.unwrap_or(0),
))
})
.collect::<Vec<_>>()
@ -1297,7 +1385,9 @@ async fn enforce_cache_limit(state: &EngineState) {
}
}
cancel_torrent_root(state, torrent_id);
debug_log(format!("[TorrServer] evicted inactive torrent={torrent_id} for cache limit"));
debug_log(format!(
"[TorrServer] evicted inactive torrent={torrent_id} for cache limit"
));
}
}
}
@ -1331,13 +1421,17 @@ fn playback_session_for(
.ok()
.and_then(|windows| windows.get(&key).map(|window| window.playback_offset));
if let Ok(mut sessions) = state.playback_sessions.lock() {
let seek = previous_offset.is_some_and(|previous| offset.abs_diff(previous) > seek_threshold);
let seek =
previous_offset.is_some_and(|previous| offset.abs_diff(previous) > seek_threshold);
if seek {
if let Some(previous) = sessions.get(&key) {
previous.cancel.cancel();
}
}
let generation = sessions.get(&key).map(|session| session.generation).unwrap_or(0)
let generation = sessions
.get(&key)
.map(|session| session.generation)
.unwrap_or(0)
+ u64::from(seek || !sessions.contains_key(&key));
let session = sessions.entry(key).or_insert_with(|| PlaybackSession {
generation,
@ -1393,7 +1487,12 @@ fn is_probe_range(
is_probe_for_window(window, offset, length, MAX_PROBE_BYTES)
}
fn is_probe_for_window(window: PlaybackWindow, offset: u64, length: u64, max_probe_bytes: u64) -> bool {
fn is_probe_for_window(
window: PlaybackWindow,
offset: u64,
length: u64,
max_probe_bytes: u64,
) -> bool {
length <= max_probe_bytes
&& offset.abs_diff(window.playback_offset)
> (window.warm_ahead_bytes / 4).max(max_probe_bytes)
@ -1417,7 +1516,9 @@ fn playback_phase(
Some(TorrentStatsState::Paused) => "stalled",
Some(TorrentStatsState::Live)
if window.is_some_and(|window| {
window.seek_started_at.is_some_and(|started| started.elapsed() < Duration::from_secs(2))
window
.seek_started_at
.is_some_and(|started| started.elapsed() < Duration::from_secs(2))
}) =>
{
"seeking"
@ -1528,8 +1629,18 @@ mod tests {
seek_started_at: None,
updated_at: std::time::Instant::now(),
};
assert!(is_probe_for_window(window, 500 * 1024 * 1024, 1024, 2 * 1024 * 1024));
assert!(!is_probe_for_window(window, 500 * 1024 * 1024, 8 * 1024 * 1024, 2 * 1024 * 1024));
assert!(is_probe_for_window(
window,
500 * 1024 * 1024,
1024,
2 * 1024 * 1024
));
assert!(!is_probe_for_window(
window,
500 * 1024 * 1024,
8 * 1024 * 1024,
2 * 1024 * 1024
));
}
#[test]

62
src/app_state.rs
View file

@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Mutex, OnceLock};
use std::sync::{Arc, Mutex, OnceLock};
#[derive(Clone, Debug, Default, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
@ -255,25 +255,31 @@ struct AppCoreAction {
}
static NEXT_HANDLE: AtomicU64 = AtomicU64::new(1);
static STORE: OnceLock<Mutex<HashMap<u64, AppCoreState>>> = OnceLock::new();
static STORE: OnceLock<Mutex<HashMap<u64, Arc<Mutex<AppCoreState>>>>> = OnceLock::new();
fn store() -> &'static Mutex<HashMap<u64, AppCoreState>> {
fn store() -> &'static Mutex<HashMap<u64, Arc<Mutex<AppCoreState>>>> {
STORE.get_or_init(|| Mutex::new(HashMap::new()))
}
// See headless_engine::lock_engines — recovering from poison keeps this store
// usable after a single caught panic instead of going dark for every handle.
fn lock_store() -> std::sync::MutexGuard<'static, HashMap<u64, AppCoreState>> {
fn lock_store() -> std::sync::MutexGuard<'static, HashMap<u64, Arc<Mutex<AppCoreState>>>> {
store()
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
pub fn create_app_core_state(initial_json: &str) -> u64 {
let state = serde_json::from_str(initial_json).unwrap_or_default();
let state = match serde_json::from_str(initial_json) {
Ok(state) => state,
Err(error) => {
crate::log_sink::record("create_app_core_state", &error.to_string());
return 0;
}
};
let mut states = lock_store();
let handle = NEXT_HANDLE.fetch_add(1, Ordering::Relaxed);
states.insert(handle, state);
states.insert(handle, Arc::new(Mutex::new(state)));
handle
}
@ -282,21 +288,42 @@ pub fn destroy_app_core_state(handle: u64) -> bool {
}
pub fn app_core_state_json(handle: u64) -> Option<String> {
let states = lock_store();
states
.get(&handle)
.and_then(|state| serde_json::to_string(state).ok())
let state = lock_store().get(&handle)?.clone();
let snapshot = lock_app_state(&state)?.clone();
serde_json::to_string(&snapshot).ok()
}
pub fn app_core_dispatch_json(handle: u64, action_json: &str) -> Option<String> {
let action: AppCoreAction = serde_json::from_str(action_json).ok()?;
let mut states = lock_store();
let state = states.get_mut(&handle)?;
reduce(state, action);
serde_json::to_string(state).ok()
let action: AppCoreAction = serde_json::from_str(action_json)
.map_err(|error| {
crate::log_sink::record("app_core_dispatch_json", &error.to_string());
})
.ok()?;
let state = lock_store().get(&handle)?.clone();
let snapshot = {
let mut state = lock_app_state(&state)?;
if !reduce(&mut state, action) {
crate::log_sink::record("app_core_dispatch_json", "unknown action");
return None;
}
state.clone()
};
serde_json::to_string(&snapshot).ok()
}
fn reduce(state: &mut AppCoreState, action: AppCoreAction) {
fn lock_app_state(
state: &Arc<Mutex<AppCoreState>>,
) -> Option<std::sync::MutexGuard<'_, AppCoreState>> {
match state.lock() {
Ok(guard) => Some(guard),
Err(_) => {
crate::log_sink::record("app_core_state", "poisoned handle; recreate the app state");
None
}
}
}
fn reduce(state: &mut AppCoreState, action: AppCoreAction) -> bool {
match action.action_type.as_str() {
"setHomeCategories" => state.home.categories = array_or_empty(action.value),
"setHomeLoading" => state.home.is_loading = action.value.as_bool().unwrap_or(false),
@ -347,8 +374,9 @@ fn reduce(state: &mut AppCoreState, action: AppCoreAction) {
"setCalendarLoading" => state.calendar.is_loading = action.value.as_bool().unwrap_or(false),
"setLibraryUiState" => state.library.ui_state = action.value,
"playerResetForEpisode" => reset_player_for_episode(&mut state.player, action.video_id),
_ => {}
_ => return false,
}
true
}
fn array_or_empty(value: Value) -> Value {

17
src/core_contract.rs
View file

@ -1,4 +1,5 @@
use crate::headless_engine;
use crate::runtime::EffectKind;
use serde::{Deserialize, Serialize};
use serde_json::Value;
@ -143,6 +144,17 @@ pub fn core_capabilities_json(portable: bool) -> String {
serde_json::to_string(&capabilities).unwrap_or_else(|_| "{}".to_string())
}
/// Machine-readable source for generated bindings, docs and contract drift tests.
pub fn core_contract_manifest_json() -> String {
serde_json::json!({
"version": env!("CARGO_PKG_VERSION"),
"methods": ["engine.create", "engine.snapshot", "engine.dispatch", "engine.completeEffect", "engine.destroy", "app.create", "app.state", "app.dispatch", "app.destroy"],
"effectEnvelope": {"required": ["id", "type", "generation", "payload"], "optional": ["groupId", "priority", "dedupeKey", "cachePolicy", "timeoutMs"]},
"effectTypes": EffectKind::ALL.iter().map(|kind| kind.as_str()).collect::<Vec<_>>(),
"capabilities": {"native": CoreCapabilitySet::android_default(), "portable": CoreCapabilitySet::portable_minimum()},
}).to_string()
}
fn parse_dispatch_result(json: &str) -> Option<CoreDispatchResult> {
let raw = serde_json::from_str::<Value>(json).ok()?;
let state = raw.get("state").cloned().unwrap_or(Value::Null);
@ -157,7 +169,10 @@ fn parse_dispatch_result(json: &str) -> Option<CoreDispatchResult> {
})
.unwrap_or_default();
Some(CoreDispatchResult {
revision: raw.get("revision").and_then(Value::as_u64).unwrap_or_default(),
revision: raw
.get("revision")
.and_then(Value::as_u64)
.unwrap_or_default(),
state: CoreState { value: state },
effects,
})

9
src/ffi.rs
View file

@ -371,6 +371,15 @@ mod tests {
assert_eq!(destroyed["value"], json!(true));
}
#[test]
fn lifecycle_create_rejects_malformed_initial_state() {
for method in ["engine.create", "app.create"] {
let result = parse(&core_invoke(method, "{ malformed"));
assert_eq!(result["ok"], json!(false));
assert_eq!(result["error"]["kind"], json!("invalid_args"));
}
}
#[test]
fn calendar_plan_methods_route_and_compute() {
let candidates = parse(&core_invoke(

7
src/ffi/core_addon_store_routes.rs
View file

@ -8,6 +8,7 @@ pub(super) fn route_core_contract(method: &str, args_json: &str) -> Outcome {
.and_then(|o| o.get("portable").and_then(Value::as_bool))
.unwrap_or(false),
)),
"coreContractManifest" => into_json(core_contract::core_contract_manifest_json()),
_ => Err(fail(
ErrorKind::UnknownMethod,
@ -80,9 +81,9 @@ pub(super) fn route_profile_avatar_pack(method: &str, args_json: &str) -> Outcom
// args_json IS the request object for all of these. The platform owns
// the HTTP calls between plans; this crate only validates and maps the
// GitHub responses into the stable UI contract.
"profileAvatarPackManifestPlan" => {
opt_json(profile_avatar_pack::profile_avatar_pack_manifest_plan_json(args_json))
}
"profileAvatarPackManifestPlan" => opt_json(
profile_avatar_pack::profile_avatar_pack_manifest_plan_json(args_json),
),
"profileAvatarPackRepositoryPlan" => {
opt_json(profile_avatar_pack::profile_avatar_pack_repository_plan_json(args_json))
}

20
src/ffi/engine_routes.rs
View file

@ -2,9 +2,14 @@ use super::*;
pub(super) fn route_engine_lifecycle(method: &str, args_json: &str) -> Outcome {
match method {
"engine.create" => Ok(json!(
headless_engine::create_headless_engine(args_json) as i64
)),
"engine.create" => {
let handle = headless_engine::create_headless_engine(args_json);
if handle == 0 {
Err(fail(ErrorKind::InvalidArgs, "invalid initial engine state"))
} else {
Ok(json!(handle as i64))
}
}
"engine.snapshot" => result_json(
headless_engine::headless_engine_snapshot_json(handle(args_json)?),
method,
@ -33,7 +38,14 @@ pub(super) fn route_engine_lifecycle(method: &str, args_json: &str) -> Outcome {
args_json
)?))),
"core.drainErrorLog" => opt_json(Some(crate::log_sink::drain_core_log_json())),
"app.create" => Ok(json!(app_state::create_app_core_state(args_json) as i64)),
"app.create" => {
let handle = app_state::create_app_core_state(args_json);
if handle == 0 {
Err(fail(ErrorKind::InvalidArgs, "invalid initial app state"))
} else {
Ok(json!(handle as i64))
}
}
"app.state" => result_json(app_state::app_core_state_json(handle(args_json)?), method),
"app.dispatch" => {
let args = object(args_json)?;

22
src/ffi/intro_plugins_routes.rs
View file

@ -9,9 +9,9 @@ pub(super) fn route_intro_segments(method: &str, args_json: &str) -> Outcome {
"skipdbSegmentsPlan" => opt_json(intro_segments::skipdb_segments_plan_json(args_json)),
"skipdbSubmitPlan" => opt_json(intro_segments::skipdb_submit_plan_json(args_json)),
"parseSkipdbSegments" => opt_json(intro_segments::parse_skipdb_segments_json(args_json)),
"parsePublicmetadbSegments" => opt_json(
intro_segments::parse_publicmetadb_segments_json(args_json),
),
"parsePublicmetadbSegments" => {
opt_json(intro_segments::parse_publicmetadb_segments_json(args_json))
}
"anilistMalId" => opt_json(intro_segments::anilist_mal_id_json(args_json)),
"anilistId" => opt_json(intro_segments::anilist_id_json(args_json)),
"anilistMediaIdPlan" => opt_json(intro_segments::anilist_media_id_plan_json(args_json)),
@ -21,12 +21,12 @@ pub(super) fn route_intro_segments(method: &str, args_json: &str) -> Outcome {
opt_json(intro_segments::anime_skip_find_show_plan_json(args_json))
}
"animeSkipShowId" => opt_json(intro_segments::anime_skip_show_id_json(args_json)),
"animeSkipFindEpisodesPlan" => {
opt_json(intro_segments::anime_skip_find_episodes_plan_json(args_json))
}
"animeSkipFindTimestampsPlan" => {
opt_json(intro_segments::anime_skip_find_timestamps_plan_json(args_json))
}
"animeSkipFindEpisodesPlan" => opt_json(
intro_segments::anime_skip_find_episodes_plan_json(args_json),
),
"animeSkipFindTimestampsPlan" => opt_json(
intro_segments::anime_skip_find_timestamps_plan_json(args_json),
),
"parseAnimeSkipResults" => {
opt_json(intro_segments::parse_anime_skip_results_json(args_json))
}
@ -34,9 +34,7 @@ pub(super) fn route_intro_segments(method: &str, args_json: &str) -> Outcome {
"parseTheIntroDbSegments" => {
opt_json(intro_segments::parse_the_introdb_segments_json(args_json))
}
"theIntroDbSubmitPlan" => {
opt_json(intro_segments::the_introdb_submit_plan_json(args_json))
}
"theIntroDbSubmitPlan" => opt_json(intro_segments::the_introdb_submit_plan_json(args_json)),
"uniqueIntroSegments" => {
let args = object(args_json)?;
opt_json(intro_segments::unique_intro_segments_json(

21
src/headless_engine/complete_effect.rs
View file

@ -8,25 +8,20 @@ use crate::runtime::{EffectEnvelope, EffectKind};
impl HeadlessEngine {
pub(super) fn complete_effect(&mut self, result: EffectResultInput) -> Vec<EffectEnvelope> {
let Some(effect) = self
.state
.pending_effects
.iter()
.find(|effect| effect.id == result.effect_id)
.cloned()
else {
let Some(effect) = self.take_pending_effect(&result.effect_id) else {
return vec![];
};
let generation = effect.generation;
// Unknown effect type (e.g. stale build mismatch between platform and core) — drop silently.
// A stale build mismatch still needs to clear the runtime registry entry. The FFI
// boundary has no structured logging facility yet, so keep this non-fatal while making
// the mismatch observable to hosts that collect stderr.
let Some(kind) = EffectKind::from_str(&effect.kind) else {
eprintln!(
"fluxa-core contract mismatch: completion for unknown effect type '{}' (id '{}', generation {})",
effect.kind, effect.id, effect.generation
);
return vec![];
};
self.state
.pending_effects
.retain(|pending| pending.id != result.effect_id);
self.delivered_effect_ids.remove(&result.effect_id);
self.effect_created_at.remove(&result.effect_id);
let effect_type = kind.as_str();
// No wildcard arm: adding an EffectKind variant without handling it here is a compile error.

2
src/headless_engine/contracts.rs
View file

@ -400,6 +400,4 @@ pub(super) struct StatePatch {
pub trailer: Option<super::trailer::TrailerState>,
#[serde(skip_serializing_if = "Option::is_none")]
pub plugins: Option<super::plugins::PluginsState>,
#[serde(skip_serializing_if = "Option::is_none")]
pub pending_effects: Option<Vec<EffectEnvelope>>,
}

73
src/headless_engine/effect_bookkeeping.rs
View file

@ -13,7 +13,11 @@ impl HeadlessEngine {
payload: P,
) -> EffectEnvelope {
let payload = serde_json::to_value(&payload).unwrap_or(Value::Null);
self.effect_raw(kind.as_str(), generation, payload)
let id = format!("fx-{}", self.next_effect_id);
self.next_effect_id += 1;
let envelope = EffectEnvelope::new(id.clone(), kind, generation, payload);
self.register_effect(id, envelope.clone());
envelope
}
// For pass-through of effects emitted by sub-modules (e.g. player_flow) where
@ -27,17 +31,31 @@ impl HeadlessEngine {
let id = format!("fx-{}", self.next_effect_id);
self.next_effect_id += 1;
let envelope = EffectEnvelope::raw(id.clone(), kind, generation, payload);
self.state.pending_effects.push(envelope.clone());
self.effect_created_at.insert(id, Instant::now());
self.register_effect(id, envelope.clone());
envelope
}
fn register_effect(&mut self, id: String, envelope: EffectEnvelope) {
self.pending_effects.push(envelope);
self.effect_created_at.insert(id, Instant::now());
}
pub(super) fn take_pending_effect(&mut self, id: &str) -> Option<EffectEnvelope> {
let index = self
.pending_effects
.iter()
.position(|effect| effect.id == id)?;
let effect = self.pending_effects.remove(index);
self.delivered_effect_ids.remove(id);
self.effect_created_at.remove(id);
Some(effect)
}
// Drops any pending effect old enough that it's almost certainly been abandoned by
// the platform rather than genuinely still in flight. Called opportunistically on
// every dispatch/complete_effect so no background timer is needed.
pub(super) fn expire_stale_pending_effects(&mut self, now: Instant) {
let stale_ids: Vec<String> = self
.state
.pending_effects
.iter()
.filter(|effect| {
@ -48,7 +66,7 @@ impl HeadlessEngine {
.map(|effect| effect.id.clone())
.collect();
for id in &stale_ids {
self.state.pending_effects.retain(|effect| &effect.id != id);
self.pending_effects.retain(|effect| &effect.id != id);
self.delivered_effect_ids.remove(id);
self.effect_created_at.remove(id);
}
@ -80,11 +98,52 @@ impl HeadlessEngine {
}
pub(super) fn undelivered_pending_effects(&self) -> Vec<EffectEnvelope> {
self.state
.pending_effects
self.pending_effects
.iter()
.filter(|effect| !self.delivered_effect_ids.contains(&effect.id))
.cloned()
.collect()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::headless_engine::contracts::{EffectResultInput, EffectStatus};
use serde_json::json;
#[test]
fn typed_effects_use_their_schedule_metadata() {
let mut engine = HeadlessEngine::default();
let effect = engine.effect(
EffectKind::FetchCatalogPage,
3,
json!({ "url": "https://example.com/catalog" }),
);
assert_eq!(effect.group_id.as_deref(), Some("addon"));
assert_eq!(effect.priority, 50);
assert_eq!(effect.cache_policy.as_deref(), Some("default"));
assert_eq!(effect.timeout_ms, Some(15_000));
assert!(effect.dedupe_key.is_some());
}
#[test]
fn unknown_effect_completion_removes_the_runtime_entry() {
let mut engine = HeadlessEngine::default();
let effect = engine.effect_raw("effectFromNewerCore", 1, json!({}));
let effects = engine.complete_effect(EffectResultInput {
effect_id: effect.id,
status: EffectStatus::Ok,
value: serde_json::Value::Null,
error: serde_json::Value::Null,
});
assert!(effects.is_empty());
assert!(engine.pending_effects.is_empty());
assert!(engine.delivered_effect_ids.is_empty());
assert!(engine.effect_created_at.is_empty());
}
}

100
src/headless_engine/mod.rs
View file

@ -26,11 +26,10 @@ mod youtube_cipher;
use crate::core_error::{CoreError, LogAndDiscard};
use crate::runtime::EffectEnvelope;
use contracts::{AppAction, DispatchResult, StatePatch};
use serde::{Deserialize, Serialize};
use state::EngineState;
use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Mutex, OnceLock};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Duration;
use web_time::Instant;
@ -42,46 +41,44 @@ pub(crate) use contracts::EffectResultInput;
// Anything genuinely still in flight completes well within this window.
const EFFECT_EXPIRY: Duration = Duration::from_secs(300);
#[derive(Clone, Debug, Default, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
#[derive(Debug, Default)]
struct HeadlessEngine {
#[serde(default)]
state: EngineState,
#[serde(default = "first_effect_id")]
next_effect_id: u64,
#[serde(default)]
revision: u64,
// Ids handed to the platform at least once, awaiting their complete_effect call.
// Never serialized — purely tracks delivery so the "drain the queue" fallback in
// resolve_visible_effects doesn't hand out an effect that's already in flight as
// if it were fresh work (which used to make an unrelated dispatch while a slow
// effect was still running re-trigger a full duplicate execution of it).
#[serde(skip)]
delivered_effect_ids: HashSet<String>,
// When each pending effect was created, for expire_stale_pending_effects. Never
// serialized — Instant isn't a portable wall-clock value, just an internal timer.
#[serde(skip)]
effect_created_at: HashMap<String, Instant>,
}
fn first_effect_id() -> u64 {
1
// Runtime-only effect registry. Effect payloads can contain credentials and must never
// be included in a UI state snapshot or StatePatch.
pending_effects: Vec<EffectEnvelope>,
}
static ENGINE_COUNTER: AtomicU64 = AtomicU64::new(1);
static ENGINES: OnceLock<Mutex<HashMap<u64, HeadlessEngine>>> = OnceLock::new();
static ENGINES: OnceLock<Mutex<HashMap<u64, Arc<Mutex<HeadlessEngine>>>>> = OnceLock::new();
pub fn create_headless_engine(initial_json: &str) -> u64 {
let mut engine = HeadlessEngine {
next_effect_id: 1,
..HeadlessEngine::default()
};
if let Ok(initial_state) = serde_json::from_str::<EngineState>(initial_json) {
engine.state = initial_state;
}
let initial_state = match serde_json::from_str::<EngineState>(initial_json) {
Ok(state) => state,
Err(error) => {
crate::log_sink::record("create_headless_engine", &error.to_string());
return 0;
}
};
engine.state = initial_state;
let mut map = lock_engines();
let handle = ENGINE_COUNTER.fetch_add(1, Ordering::Relaxed);
map.insert(handle, engine);
map.insert(handle, Arc::new(Mutex::new(engine)));
handle
}
@ -92,8 +89,9 @@ pub fn destroy_headless_engine(handle: u64) -> bool {
pub fn headless_engine_snapshot_json(handle: u64) -> Option<String> {
let state = {
let map = lock_engines();
map.get(&handle)?.state.clone()
map.get(&handle)?.clone()
};
let state = lock_engine(&state)?.state.clone();
serde_json::to_string(&state).ok()
}
@ -105,9 +103,8 @@ pub fn headless_engine_dispatch_json(handle: u64, action_json: &str) -> Option<S
})
.log_discard()?;
let (revision, patch, visible_effects) = {
let mut map = lock_engines();
let engine = match map.get_mut(&handle) {
Some(engine) => engine,
let engine = match lock_engines().get(&handle) {
Some(engine) => Arc::clone(engine),
None => {
return CoreError::NotFound {
context: "headless_engine_dispatch_json",
@ -115,6 +112,12 @@ pub fn headless_engine_dispatch_json(handle: u64, action_json: &str) -> Option<S
.log_and_none();
}
};
let mut engine = lock_engine(&engine).or_else(|| {
CoreError::NotFound {
context: "headless_engine_dispatch_json (poisoned handle)",
}
.log_and_none()
})?;
engine.expire_stale_pending_effects(Instant::now());
let effects = engine.dispatch(action);
let visible_effects = engine.resolve_visible_effects(effects);
@ -132,9 +135,8 @@ pub fn headless_engine_complete_effect_json(handle: u64, result_json: &str) -> O
})
.log_discard()?;
let (revision, patch, visible_effects) = {
let mut map = lock_engines();
let engine = match map.get_mut(&handle) {
Some(engine) => engine,
let engine = match lock_engines().get(&handle) {
Some(engine) => Arc::clone(engine),
None => {
return CoreError::NotFound {
context: "headless_engine_complete_effect_json",
@ -142,6 +144,12 @@ pub fn headless_engine_complete_effect_json(handle: u64, result_json: &str) -> O
.log_and_none();
}
};
let mut engine = lock_engine(&engine).or_else(|| {
CoreError::NotFound {
context: "headless_engine_complete_effect_json (poisoned handle)",
}
.log_and_none()
})?;
engine.expire_stale_pending_effects(Instant::now());
let effects = engine.complete_effect(result);
let visible_effects = engine.resolve_visible_effects(effects);
@ -153,23 +161,41 @@ pub fn headless_engine_complete_effect_json(handle: u64, result_json: &str) -> O
// Deliberately takes owned before/after snapshots rather than a reference to the locked
// engine: diffing and serializing a large state (e.g. a big discover catalog) can take
// over a second, and every other Tauri command shares one global engine mutex — holding
// it for that long would stall unrelated IPC calls behind it. Callers clone what they
// need and drop the lock before calling this.
fn result_patch_json(revision: u64, state: StatePatch, effects: Vec<EffectEnvelope>) -> Option<String> {
serde_json::to_string(&DispatchResult { revision, state, effects }).ok()
// over a second. Callers clone what they need and drop the per-engine lock before calling
// this, so unrelated engine handles continue to make progress.
fn result_patch_json(
revision: u64,
state: StatePatch,
effects: Vec<EffectEnvelope>,
) -> Option<String> {
serde_json::to_string(&DispatchResult {
revision,
state,
effects,
})
.ok()
}
fn engines() -> &'static Mutex<HashMap<u64, HeadlessEngine>> {
fn engines() -> &'static Mutex<HashMap<u64, Arc<Mutex<HeadlessEngine>>>> {
ENGINES.get_or_init(|| Mutex::new(HashMap::new()))
}
// A panic while a request held this lock poisons it; with catch_unwind now
// guarding the FFI boundary, a single caught panic must not silently make
// every engine handle inaccessible for the rest of the process's life.
// Recovering the guard accepts that one engine's state might be left
// mid-update, which is still far better than every other handle going dark.
fn lock_engines() -> std::sync::MutexGuard<'static, HashMap<u64, HeadlessEngine>> {
fn lock_engine(
engine: &Arc<Mutex<HeadlessEngine>>,
) -> Option<std::sync::MutexGuard<'_, HeadlessEngine>> {
match engine.lock() {
Ok(guard) => Some(guard),
Err(_) => {
crate::log_sink::record("headless_engine", "poisoned handle; recreate the engine");
None
}
}
}
// A panic while a request held the registry lock poisons it; recover so a caught panic
// does not make every handle inaccessible. A poisoned engine itself is isolated to its
// own per-handle lock.
fn lock_engines() -> std::sync::MutexGuard<'static, HashMap<u64, Arc<Mutex<HeadlessEngine>>>> {
engines()
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())

3
src/headless_engine/state.rs
View file

@ -14,7 +14,6 @@ use super::search::SearchState;
use super::settings::SettingsState;
use super::sync::SyncState;
use super::trailer::TrailerState;
use crate::runtime::EffectEnvelope;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use std::ops::{Deref, DerefMut};
@ -205,7 +204,6 @@ pub(super) struct EngineState {
pub(super) offline: Tracked<OfflineState>,
pub(super) trailer: Tracked<TrailerState>,
pub(super) plugins: Tracked<PluginsState>,
pub(super) pending_effects: Tracked<Vec<EffectEnvelope>>,
#[serde(rename = "_runtime")]
pub(super) runtime: RuntimeGenerations,
}
@ -230,7 +228,6 @@ impl EngineState {
offline: self.offline.take_if_dirty(),
trailer: self.trailer.take_if_dirty(),
plugins: self.plugins.take_if_dirty(),
pending_effects: self.pending_effects.take_if_dirty(),
}
}
}

17
src/headless_engine/tests/engine.rs
View file

@ -17,3 +17,20 @@ fn engines_lock_survives_a_panic_while_held_by_another_thread() {
assert!(headless_engine_snapshot_json(handle).is_some());
assert!(destroy_headless_engine(handle));
}
#[test]
fn poisoned_engine_handle_is_rejected_without_affecting_other_handles() {
let poisoned_handle = create_headless_engine("{}");
let healthy_handle = create_headless_engine("{}");
let poisoned_engine = lock_engines().get(&poisoned_handle).unwrap().clone();
let poisoner = std::thread::spawn(move || {
let _guard = poisoned_engine.lock().unwrap();
panic!("simulated engine update panic");
});
assert!(poisoner.join().is_err());
assert!(headless_engine_snapshot_json(poisoned_handle).is_none());
assert!(headless_engine_snapshot_json(healthy_handle).is_some());
assert!(destroy_headless_engine(poisoned_handle));
assert!(destroy_headless_engine(healthy_handle));
}

4
src/headless_engine/tests/library.rs
View file

@ -201,13 +201,13 @@ fn expire_stale_pending_effects_drops_old_but_not_recent_effects() {
// Still well within the window — nothing genuinely in flight should be dropped.
engine.expire_stale_pending_effects(Instant::now());
assert_eq!(engine.state.pending_effects.len(), 2);
assert_eq!(engine.pending_effects.len(), 2);
// Past the expiry window — abandoned effects (platform never called
// complete_effect) get swept from all three bookkeeping collections.
let far_future = Instant::now() + Duration::from_secs(301);
engine.expire_stale_pending_effects(far_future);
assert!(engine.state.pending_effects.is_empty());
assert!(engine.pending_effects.is_empty());
assert!(engine.delivered_effect_ids.is_empty());
assert!(engine.effect_created_at.is_empty());
}

70
src/intro_segments.rs
View file

@ -1,7 +1,12 @@
use serde_json::{Value, json};
use std::collections::HashMap;
fn request_plan(url: String, method: &str, headers: &[(&str, String)], body: Option<&Value>) -> Value {
fn request_plan(
url: String,
method: &str,
headers: &[(&str, String)],
body: Option<&Value>,
) -> Value {
let headers_obj: serde_json::Map<String, Value> = headers
.iter()
.map(|(k, v)| (k.to_string(), Value::String(v.clone())))
@ -19,7 +24,9 @@ pub(crate) fn intro_db_segments_plan_json(args_json: &str) -> Option<String> {
let imdb_id = args.get("imdbId").and_then(Value::as_str)?;
let season = args.get("season").and_then(Value::as_i64)?;
let episode = args.get("episode").and_then(Value::as_i64)?;
let url = format!("https://api.introdb.app/segments?imdb_id={imdb_id}&season={season}&episode={episode}");
let url = format!(
"https://api.introdb.app/segments?imdb_id={imdb_id}&season={season}&episode={episode}"
);
serde_json::to_string(&request_plan(url, "GET", &[], None)).ok()
}
@ -64,7 +71,9 @@ pub(crate) fn skipdb_segments_plan_json(args_json: &str) -> Option<String> {
let imdb_id = args.get("imdbId").and_then(Value::as_str)?;
let season = args.get("season").and_then(Value::as_i64)?;
let episode = args.get("episode").and_then(Value::as_i64)?;
let url = format!("https://api.skipdb.tv/api/segments?imdb_id={imdb_id}&season={season}&episode={episode}");
let url = format!(
"https://api.skipdb.tv/api/segments?imdb_id={imdb_id}&season={season}&episode={episode}"
);
serde_json::to_string(&request_plan(url, "GET", &[], None)).ok()
}
@ -135,7 +144,10 @@ fn clean_anilist_title(title: &str) -> String {
let trimmed = title.trim_end();
if let Some(idx) = trimmed.rfind(" (") {
let inner = &trimmed[idx + 2..];
if trimmed.ends_with(')') && inner.len() == 5 && inner[..4].chars().all(|c| c.is_ascii_digit()) {
if trimmed.ends_with(')')
&& inner.len() == 5
&& inner[..4].chars().all(|c| c.is_ascii_digit())
{
return trimmed[..idx].trim().to_string();
}
}
@ -603,10 +615,22 @@ fn the_introdb_canonical_type(wire: &str) -> &'static str {
pub(crate) fn the_introdb_media_plan_json(args_json: &str) -> Option<String> {
let args: Value = serde_json::from_str(args_json).ok()?;
let tmdb_id = args.get("tmdbId").and_then(Value::as_i64);
let imdb_id = args.get("imdbId").and_then(Value::as_str).filter(|s| s.starts_with("tt"));
let season = args.get("season").and_then(Value::as_i64).filter(|s| *s > 0);
let episode = args.get("episode").and_then(Value::as_i64).filter(|e| *e > 0);
let duration_ms = args.get("durationMs").and_then(Value::as_i64).filter(|d| *d > 0);
let imdb_id = args
.get("imdbId")
.and_then(Value::as_str)
.filter(|s| s.starts_with("tt"));
let season = args
.get("season")
.and_then(Value::as_i64)
.filter(|s| *s > 0);
let episode = args
.get("episode")
.and_then(Value::as_i64)
.filter(|e| *e > 0);
let duration_ms = args
.get("durationMs")
.and_then(Value::as_i64)
.filter(|d| *d > 0);
let mut query = match (tmdb_id, imdb_id) {
(Some(id), _) => format!("tmdb_id={id}"),
@ -625,12 +649,18 @@ pub(crate) fn the_introdb_media_plan_json(args_json: &str) -> Option<String> {
pub(crate) fn parse_the_introdb_segments_json(args_json: &str) -> Option<String> {
let args: Value = serde_json::from_str(args_json).ok()?;
let response: Value = serde_json::from_str(args.get("responseJson").and_then(Value::as_str)?).ok()?;
let duration_ms = args.get("durationMs").and_then(Value::as_i64).filter(|d| *d > 0);
let response: Value =
serde_json::from_str(args.get("responseJson").and_then(Value::as_str)?).ok()?;
let duration_ms = args
.get("durationMs")
.and_then(Value::as_i64)
.filter(|d| *d > 0);
let mut segments = Vec::new();
for wire_type in &["intro", "recap", "credits", "preview"] {
let Some(items) = response.get(*wire_type).and_then(Value::as_array) else { continue };
let Some(items) = response.get(*wire_type).and_then(Value::as_array) else {
continue;
};
let canonical = the_introdb_canonical_type(wire_type);
for item in items {
let start_ms = item.get("start_ms").and_then(Value::as_i64).unwrap_or(0);
@ -660,8 +690,14 @@ pub(crate) fn the_introdb_submit_plan_json(args_json: &str) -> Option<String> {
let end_sec = args.get("endSec").and_then(Value::as_f64);
let video_duration_ms = args.get("videoDurationMs").and_then(Value::as_i64);
let imdb_id = args.get("imdbId").and_then(Value::as_str);
let season = args.get("season").and_then(Value::as_i64).filter(|s| *s > 0);
let episode = args.get("episode").and_then(Value::as_i64).filter(|e| *e > 0);
let season = args
.get("season")
.and_then(Value::as_i64)
.filter(|s| *s > 0);
let episode = args
.get("episode")
.and_then(Value::as_i64)
.filter(|e| *e > 0);
let mut body = json!({
"tmdb_id": tmdb_id,
@ -727,8 +763,10 @@ mod tests {
.iter()
.any(|s| s["type"] == "intro" && s["startTime"] == 0 && s["endTime"] == 62000)
);
assert!(segments.iter().any(|s| s["type"] == "outro"
&& s["startTime"] == 3180000
&& s["endTime"] == 3240000));
assert!(
segments.iter().any(|s| s["type"] == "outro"
&& s["startTime"] == 3180000
&& s["endTime"] == 3240000)
);
}
}

26
src/plugin_runtime/host_functions.rs
View file

@ -1,5 +1,8 @@
use super::dom_bridge::DomBridge;
use super::{PluginHttpClient, PluginHttpRequest, crypto_bridge};
use super::{
PLUGIN_MAX_RESPONSE_BODY_BYTES, PluginHttpClient, PluginHttpRequest, crypto_bridge,
plugin_http_request_error,
};
use rquickjs::{Ctx, Function};
use std::collections::HashMap;
use std::rc::Rc;
@ -14,18 +17,31 @@ fn native_fetch(
follow_redirects: bool,
) -> String {
let headers: HashMap<String, String> = serde_json::from_str(&headers_json).unwrap_or_default();
let response = client.fetch(PluginHttpRequest {
let request = PluginHttpRequest {
method,
url,
headers,
body,
follow_redirects,
});
};
if let Some(error) = plugin_http_request_error(&request) {
return format!(
"{{\"ok\":false,\"status\":0,\"body\":\"\",\"error\":{}}}",
serde_json::to_string(error).unwrap_or_else(|_| "\"plugin request rejected\"".into())
);
}
let mut response = client.fetch(request);
if response.body.len() > PLUGIN_MAX_RESPONSE_BODY_BYTES {
response.body.truncate(PLUGIN_MAX_RESPONSE_BODY_BYTES);
response.ok = false;
response.error = Some("plugin response body exceeds limit".to_string());
}
format!(
"{{\"ok\":{},\"status\":{},\"body\":{}}}",
"{{\"ok\":{},\"status\":{},\"body\":{},\"error\":{}}}",
response.ok,
response.status,
serde_json::to_string(&response.body).unwrap_or_else(|_| "\"\"".into())
serde_json::to_string(&response.body).unwrap_or_else(|_| "\"\"".into()),
serde_json::to_string(&response.error).unwrap_or_else(|_| "null".into())
)
}

65
src/plugin_runtime/mod.rs
View file

@ -13,6 +13,57 @@ use std::time::Duration;
pub(super) const PLUGIN_TIMEOUT_SECS: u64 = 60;
pub(super) const PLUGIN_MEMORY_LIMIT: usize = 256 * 1024 * 1024;
pub const PLUGIN_MAX_REQUEST_BODY_BYTES: usize = 1_048_576;
pub const PLUGIN_MAX_RESPONSE_BODY_BYTES: usize = 8 * 1_048_576;
pub fn plugin_http_request_error(request: &PluginHttpRequest) -> Option<&'static str> {
let lower = request.url.to_ascii_lowercase();
if !(lower.starts_with("http://") || lower.starts_with("https://")) {
return Some("only http and https plugin URLs are allowed");
}
let authority = lower
.split("//")
.nth(1)?
.split('/')
.next()?
.split('@')
.next_back()?;
let host = authority.split(':').next().unwrap_or(authority);
let private_172 = host
.strip_prefix("172.")
.and_then(|rest| rest.split('.').next())
.and_then(|part| part.parse::<u8>().ok())
.is_some_and(|second| (16..=31).contains(&second));
if host == "localhost"
|| host.ends_with(".localhost")
|| host == "::1"
|| host.starts_with("127.")
|| host.starts_with("10.")
|| host.starts_with("192.168.")
|| host.starts_with("169.254.")
|| host.starts_with("0.")
|| private_172
|| host.starts_with("fc")
|| host.starts_with("fd")
|| host.starts_with("fe80:")
{
return Some("private or loopback plugin URL is not allowed");
}
if let Some(body) = &request.body
&& body.len() > PLUGIN_MAX_REQUEST_BODY_BYTES
{
return Some("plugin request body exceeds limit");
}
if request.headers.keys().any(|key| {
!matches!(
key.to_ascii_lowercase().as_str(),
"accept" | "accept-language" | "content-type" | "origin" | "referer" | "user-agent"
)
}) {
return Some("plugin request contains a disallowed header");
}
None
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[cfg_attr(feature = "uniffi-bindings", derive(uniffi::Record))]
@ -115,6 +166,20 @@ mod tests {
Arc::new(MockHttpClient)
}
#[test]
fn plugin_request_policy_rejects_private_hosts_and_credentials() {
let mut request = PluginHttpRequest {
url: "http://127.0.0.1/admin".to_string(),
..PluginHttpRequest::default()
};
assert!(plugin_http_request_error(&request).is_some());
request.url = "https://example.com/".to_string();
request
.headers
.insert("Authorization".to_string(), "secret".to_string());
assert!(plugin_http_request_error(&request).is_some());
}
fn run_scraper(code: &str, tmdb_id: &str, media_type: &str) -> String {
execute_scraper(
mock_client(),

16
src/profile_avatar_pack.rs
View file

@ -213,7 +213,10 @@ fn direct_manifest_url(input: &str) -> Option<String> {
fn is_manifest_filename(url: &str) -> bool {
let path = url.split('?').next().unwrap_or(url);
let filename = path.rsplit('/').next().unwrap_or("");
matches!(filename.to_ascii_lowercase().as_str(), "pack.json" | "json.pack")
matches!(
filename.to_ascii_lowercase().as_str(),
"pack.json" | "json.pack"
)
}
fn parse_repository_url(input: &str) -> Option<GitHubRepository> {
@ -250,8 +253,10 @@ fn extract_target_path(rest: &str) -> Option<String> {
}
let path = percent_decode(path);
let (directory, filename) = path.rsplit_once('/').unwrap_or(("", &path));
let directory = if matches!(filename.to_ascii_lowercase().as_str(), "pack.json" | "json.pack")
{
let directory = if matches!(
filename.to_ascii_lowercase().as_str(),
"pack.json" | "json.pack"
) {
directory
} else {
path.as_str()
@ -465,7 +470,10 @@ mod tests {
.unwrap(),
)
.unwrap();
assert_eq!(output["manifestUrl"], "https://example.com/packs/solo-leveling/pack.json");
assert_eq!(
output["manifestUrl"],
"https://example.com/packs/solo-leveling/pack.json"
);
}
#[test]

135
src/runtime/effects.rs
View file

@ -1,4 +1,5 @@
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
/// Exhaustive catalog of all effect types the headless engine can emit.
///
@ -60,6 +61,57 @@ pub enum EffectKind {
}
impl EffectKind {
pub const ALL: &[Self] = &[
Self::ClearPlaybackProgress,
Self::EnqueueOfflineDownload,
Self::EnqueueTraktScrobble,
Self::ExchangeAuthCode,
Self::ExecutePlugin,
Self::FetchAddonManifest,
Self::FetchAddonResource,
Self::FetchCatalogPage,
Self::FetchDiscoverPage,
Self::FetchDetailSecondary,
Self::FetchDetailStreams,
Self::FetchIntroSegments,
Self::FetchMetaDetail,
Self::FetchMetaDetailLookup,
Self::FetchPluginManifest,
Self::FetchSeasonEpisodes,
Self::FetchSubtitles,
Self::FetchYoutubeTrailerPlayer,
Self::FetchYoutubeTrailerPlayerScript,
Self::FetchYoutubeTrailerWatchConfig,
Self::LoadStreams,
Self::NotifyReleasedEpisodes,
Self::PrefetchDetailStreams,
Self::PrefetchNextEpisodeStreams,
Self::PrepareDirectPlayback,
Self::ReadCalendarMonth,
Self::ReadDetailLocalState,
Self::ReadDiscoverCatalogFilters,
Self::ReadHomeBootstrap,
Self::RefreshContinueWatching,
Self::ReadLibraryState,
Self::ReadPlaybackProgress,
Self::RefreshAuthToken,
Self::RefreshInstalledAddons,
Self::ReplaceExternalContinueWatching,
Self::ResolveIntroImdbId,
Self::RunAuthFlow,
Self::RunDiscover,
Self::RunExternalSync,
Self::RunSearch,
Self::StartTorrentStream,
Self::StopTorrent,
Self::SyncExternalIntegration,
Self::SyncWatchedState,
Self::UpdateCalendarWidget,
Self::WriteFeedback,
Self::WriteLibraryCommand,
Self::WritePlaybackProgress,
Self::WriteSettings,
];
pub fn as_str(self) -> &'static str {
match self {
EffectKind::ClearPlaybackProgress => "clearPlaybackProgress",
@ -196,7 +248,10 @@ pub struct EffectEnvelope {
pub payload: serde_json::Value,
#[serde(skip_serializing_if = "Option::is_none")]
pub group_id: Option<String>,
#[serde(default = "default_priority", skip_serializing_if = "is_default_priority")]
#[serde(
default = "default_priority",
skip_serializing_if = "is_default_priority"
)]
pub priority: u8,
#[serde(skip_serializing_if = "Option::is_none")]
pub dedupe_key: Option<String>,
@ -210,11 +265,9 @@ impl EffectEnvelope {
pub fn new(id: String, kind: EffectKind, generation: u64, payload: serde_json::Value) -> Self {
let (group_id, priority, cache_policy, timeout_ms) = effect_schedule(kind);
Self {
dedupe_key: group_id.as_ref().and_then(|_| {
serde_json::to_string(&payload)
.ok()
.map(|payload_key| format!("{}:{generation}:{payload_key}", kind.as_str()))
}),
dedupe_key: group_id
.as_ref()
.and_then(|_| payload_dedupe_key(kind, generation, &payload)),
id,
kind: kind.as_str().to_owned(),
generation,
@ -241,6 +294,20 @@ impl EffectEnvelope {
}
}
fn payload_dedupe_key(
kind: EffectKind,
generation: u64,
payload: &serde_json::Value,
) -> Option<String> {
let payload = serde_json::to_string(payload).ok()?;
let digest = Sha256::digest(payload.as_bytes());
let hash = digest
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>();
Some(format!("{}:{generation}:{hash}", kind.as_str()))
}
fn default_priority() -> u8 {
100
}
@ -256,16 +323,32 @@ fn effect_schedule(kind: EffectKind) -> (Option<String>, u8, Option<String>, Opt
| EffectKind::FetchDetailStreams
| EffectKind::PrefetchDetailStreams
| EffectKind::PrefetchNextEpisodeStreams
| EffectKind::RefreshInstalledAddons => (Some("addon".to_string()), 50, Some("default".to_string()), Some(15_000)),
EffectKind::RunSearch | EffectKind::RunDiscover => (Some("addon".to_string()), 40, Some("default".to_string()), Some(15_000)),
EffectKind::ExecutePlugin => (Some("plugin".to_string()), 60, Some("no-store".to_string()), Some(10_000)),
| EffectKind::RefreshInstalledAddons => (
Some("addon".to_string()),
50,
Some("default".to_string()),
Some(15_000),
),
EffectKind::RunSearch | EffectKind::RunDiscover => (
Some("addon".to_string()),
40,
Some("default".to_string()),
Some(15_000),
),
EffectKind::ExecutePlugin => (
Some("plugin".to_string()),
60,
Some("no-store".to_string()),
Some(10_000),
),
_ => (None, 100, None, None),
}
}
#[cfg(test)]
mod tests {
use super::EffectKind;
use super::{EffectEnvelope, EffectKind};
use serde_json::json;
#[test]
fn as_str_and_from_str_roundtrip_for_every_variant() {
@ -328,4 +411,36 @@ mod tests {
fn from_str_rejects_unknown_value() {
assert_eq!(EffectKind::from_str("notAnEffect"), None);
}
#[test]
fn scheduled_effect_dedupe_keys_hash_the_payload() {
let first = EffectEnvelope::new(
"fx-1".to_string(),
EffectKind::FetchAddonResource,
7,
json!({ "url": "https://example.com/one", "token": "secret" }),
);
let duplicate = EffectEnvelope::new(
"fx-2".to_string(),
EffectKind::FetchAddonResource,
7,
json!({ "url": "https://example.com/one", "token": "secret" }),
);
let different = EffectEnvelope::new(
"fx-3".to_string(),
EffectKind::FetchAddonResource,
7,
json!({ "url": "https://example.com/two", "token": "secret" }),
);
assert_eq!(first.dedupe_key, duplicate.dedupe_key);
assert_ne!(first.dedupe_key, different.dedupe_key);
assert!(
!first
.dedupe_key
.as_deref()
.unwrap_or_default()
.contains("secret")
);
}
}

16
src/stream_policy/torrent_runtime.rs
View file

@ -180,10 +180,22 @@ mod tests {
#[test]
fn stream_url_omits_missing_or_invalid_duration() {
let missing = build_torrent_stream_url(
"http://127.0.0.1:8090", "magnet:?xt=urn:btih:abc", "Example", None, true, false, None,
"http://127.0.0.1:8090",
"magnet:?xt=urn:btih:abc",
"Example",
None,
true,
false,
None,
);
let zero = build_torrent_stream_url(
"http://127.0.0.1:8090", "magnet:?xt=urn:btih:abc", "Example", None, true, false, Some(0),
"http://127.0.0.1:8090",
"magnet:?xt=urn:btih:abc",
"Example",
None,
true,
false,
Some(0),
);
assert!(!missing.contains("durationMs="));

25
tests/wire/expected/detail_load_requested.json
View file

@ -58,29 +58,6 @@
"userAddons": [],
"visibleStreams": [],
"watchedVideoIds": []
},
"pendingEffects": [
{
"generation": 1,
"id": "fx-1",
"payload": {
"contentType": "movie",
"id": "tt1",
"language": "en",
"profile": null,
"sourceAddonCatalogType": "",
"sourceAddonTransportUrl": ""
},
"type": "fetchMetaDetail"
},
{
"generation": 1,
"id": "fx-2",
"payload": {
"id": "tt1"
},
"type": "readPlaybackProgress"
}
]
}
}
}

18
tests/wire/expected/scrobble_requested.json
View file

@ -15,21 +15,5 @@
}
],
"revision": 1,
"state": {
"pendingEffects": [
{
"generation": 1,
"id": "fx-1",
"payload": {
"actionName": "pause",
"itemId": "tt1",
"metaType": "movie",
"profile": null,
"progress": 42.5,
"token": "trakt-token"
},
"type": "enqueueTraktScrobble"
}
]
}
"state": {}
}

11
tests/wire/expected/settings_changed.json
View file

@ -12,17 +12,6 @@
],
"revision": 1,
"state": {
"pendingEffects": [
{
"generation": 1,
"id": "fx-1",
"payload": {
"key": "subtitleLanguage",
"value": "en"
},
"type": "writeSettings"
}
],
"settings": {
"lastWriteError": null,
"values": {

20
tests/wire/expected/toggle_watchlist_requested.json
View file

@ -44,24 +44,6 @@
"savedPlaybackProgress": null,
"watched": null,
"watchlist": null
},
"pendingEffects": [
{
"generation": 1,
"id": "fx-1",
"payload": {
"command": {
"item": {
"id": "tt1",
"name": "Movie",
"type": "movie"
},
"type": "toggleWatchlist"
},
"profileId": "guest"
},
"type": "writeLibraryCommand"
}
]
}
}
}