euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-15 12:22:18 +02:00
Code Issues Packages Projects Releases 5 Wiki Activity

Feature/optimizations (#642)

Browse Source 
* - Refactored Shared-Stream hot path from Arc<Bytes> to Bytes (leveraging internal reference counting).
- Switched Metadata Trigger deduplication from Arc<Mutex<HashSet<_>>> to Arc<DashSet<_>> to reduce lock contention.
- Eliminated expensive key cloning in the InputMetadataUpdatesCompleted handler.
- Changed active_target_inputs to HashSet<Arc<str>> to avoid per-event to_string() allocations.
- Implemented bounded Cleanup-Queue (switched from unbounded to mpsc::channel(4096)) with a dedicated overflow path (try_send + async flush) to prevent uncontrolled memory growth.
- Optimized by_provider storage by switching from Vec<(addr, alloc_id)> to HashSet<(addr, alloc_id)>, reducing removal complexity from $O(n)$ to $O(1)$ average.
- Optimized ActiveConnectionInfo data layout by reordering fields (large fields first) to minimize padding and improve cache efficiency.

* Improve shared-stream reliability and align docs

- make shared connection registration fallible with rollback on failure

- guard shared broadcaster startup behind successful subscription

- optimize shared stream/provider key maps to Arc<str>

- add Tokio macros feature for runtime and dev dependencies

- harden build_resources.sh with fail-fast ffmpeg error handling

- clarify StreamError display messages with variant context

- sync README stream/custom-response sections with current behavior

* low_priority_preempted stream
* Keep custom TS timestamps monotonic across loops
- advance timestamp offset by stream duration on buffer wrap
- avoid resetting PTS/DTS to zero each cycle
- prevents players from treating looped custom streams as ended/corrupt
* low_priority_preemption custom video stream fix
* inc_version uses now main toml file
* Fixed some coderabbit issues
This commit is contained in:
euzu
2026-03-13 12:27:50 +01:00
committed by GitHub
parent 854ec93be0
commit dcbf66e201
12 changed files with 409 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+9 -9
View File
@@ -714,8 +714,8 @@ Attributes:
- `retry`, default `true`, retries the stream if it fails.
- `buffer`
- `throttle` Allowed units are `KB/s`,`MB/s`,`KiB/s`,`MiB/s`,`kbps`,`mbps`,`Mibps`. Default unit is `kbps`
- `grace_period_millis` default set to 300 milliseconds.
- `grace_period_timeout_secs` default set to 2 seconds.
- `grace_period_millis` default set to 2000 milliseconds.
- `grace_period_timeout_secs` default set to 4 seconds.
- `grace_period_hold_stream` if set to `true`, the stream will only start after the grace period check has completed. Default is `true`.
- `shared_burst_buffer_mb` optional (default `12`). Minimum burst buffer size (in MB) used for shared streams.
@@ -765,12 +765,12 @@ Default unit is `kbps`.
If you have a provider or a user where the max_connection attribute is greater than 0,
a grace period can be given during the switchover.
If this period is set too short, it may result in access being denied in some cases.
The default is 300 milliseconds.
The default is 2000 milliseconds.
If the connection is not throttled, the player will play its buffered content longer than expected.
##### 1.7.1.4 `grace_period_timeout_secs`
How long the grace grant will last, until another grace grant can made. Default is `2` seconds.
How long the grace grant will last, until another grace grant can made. Default is `4` seconds.
##### 1.7.1.5 `grace_period_hold_stream`
@@ -1057,14 +1057,14 @@ If you want to send a picture instead of black screen when a channel is not avai
Following attributes are available:
- `channel_unavailable`: _optional_
- `user_account_expired`: _optional_
- `user_connections_exhausted`: _optional_
- `provider_connections_exhausted`: _optional_
- `panel_api_provisioning`: _optional_
- `low_priority_preempted`: _optional_
- `user_account_expired`: _optional_
- `panel_api_provisioning`: _optional_
Video files with name `channel_unavailable.ts`, `user_account_expired.ts`, `user_connections_exhausted.ts`,
`provider_connections_exhausted.ts`, `panel_api_provisioning.ts`, `low_priority_preempted.ts`
Video files with name `channel_unavailable.ts`, `user_connections_exhausted.ts`, `provider_connections_exhausted.ts`,
`low_priority_preempted.ts`, `user_account_expired.ts`, `panel_api_provisioning.ts`
are already available in the docker image.
You can convert an image with `ffmpeg`.
@@ -1077,10 +1077,10 @@ and add it to the `config.yml`.
- `user_account_expired.ts`
- `provider_connections_exhausted.ts`
- `low_priority_preempted.ts`
- `user_connections_exhausted.ts`
- `channel_unavailable.ts`
- `panel_api_provisioning.ts`
- `low_priority_preempted.ts`
```yaml
custom_stream_response_path: /home/tuliprox/resources
backend/src/api/main_api.rs
+25 -34
View File
@@ -37,6 +37,7 @@ use axum::{
middleware::Next,
Router,
};
use dashmap::DashSet;
use log::{debug, error, info, warn};
use shared::{
error::TuliproxError,
@@ -58,16 +59,14 @@ const METADATA_TRIGGER_WAIT_CYCLE_LIMIT: u32 = 900;
fn collect_rescheduled_targets(
targets_set: &HashSet<String>,
running_trigger_targets: &Arc<std::sync::Mutex<HashSet<String>>>,
pending_trigger_targets: &Arc<std::sync::Mutex<HashSet<String>>>,
running_trigger_targets: &Arc<DashSet<String>>,
pending_trigger_targets: &Arc<DashSet<String>>,
) -> Vec<String> {
let mut running = running_trigger_targets.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let mut pending = pending_trigger_targets.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let mut targets_to_respawn = Vec::new();
for target in targets_set {
running.remove(target);
if pending.remove(target) {
running.insert(target.clone());
running_trigger_targets.remove(target);
if pending_trigger_targets.remove(target).is_some() {
running_trigger_targets.insert(target.clone());
targets_to_respawn.push(target.clone());
}
}
@@ -77,8 +76,8 @@ fn collect_rescheduled_targets(
fn spawn_metadata_trigger_update(
app_state: &Arc<AppState>,
targets_to_spawn: &[String],
running_trigger_targets: &Arc<std::sync::Mutex<HashSet<String>>>,
pending_trigger_targets: &Arc<std::sync::Mutex<HashSet<String>>>,
running_trigger_targets: &Arc<DashSet<String>>,
pending_trigger_targets: &Arc<DashSet<String>>,
) {
if targets_to_spawn.is_empty() {
return;
@@ -628,13 +627,13 @@ fn exec_input_update_listener(app_state: &Arc<AppState>, targets: &Arc<ProcessTa
tokio::spawn(async move {
let mut rx = app_state.event_manager.get_event_channel();
// Map<TargetName, Set<InputName>>: Tracks which inputs are currently updating for a given target
let mut active_target_inputs: HashMap<String, HashSet<String>> = HashMap::new();
let mut active_target_inputs: HashMap<String, HashSet<Arc<str>>> = HashMap::new();
// Set<TargetName>: Tracks which targets are pending an update once their inputs are done
let mut pending_targets: HashSet<String> = HashSet::new();
// Set<TargetName>: Tracks which targets have an active spawned playlist-update task (dedup guard)
let running_trigger_targets: Arc<std::sync::Mutex<HashSet<String>>> = Arc::new(std::sync::Mutex::new(HashSet::new()));
let running_trigger_targets: Arc<DashSet<String>> = Arc::new(DashSet::new());
// Set<TargetName>: Tracks follow-up triggers that arrive while a target is already running.
let pending_trigger_targets: Arc<std::sync::Mutex<HashSet<String>>> = Arc::new(std::sync::Mutex::new(HashSet::new()));
let pending_trigger_targets: Arc<DashSet<String>> = Arc::new(DashSet::new());
loop {
match rx.recv().await {
@@ -651,7 +650,7 @@ fn exec_input_update_listener(app_state: &Arc<AppState>, targets: &Arc<ProcessTa
active_target_inputs
.entry(target.name.clone())
.or_default()
.insert(input_name.to_string());
.insert(input_name.clone());
// Mark target as potentially needing an update
pending_targets.insert(target.name.clone());
}
@@ -659,37 +658,29 @@ fn exec_input_update_listener(app_state: &Arc<AppState>, targets: &Arc<ProcessTa
}
}
Ok(EventMessage::InputMetadataUpdatesCompleted(input_name)) => {
// 1. Remove this input from all active sets
let target_names: Vec<String> = active_target_inputs.keys().cloned().collect();
let mut targets_to_trigger = Vec::new();
for target_name in target_names {
if let std::collections::hash_map::Entry::Occupied(mut entry) =
active_target_inputs.entry(target_name.clone())
{
let inputs = entry.get_mut();
inputs.remove(input_name.as_ref());
if inputs.is_empty() {
entry.remove();
// If this target was pending, it's now ready to trigger
if pending_targets.remove(&target_name) {
targets_to_trigger.push(target_name);
}
// Remove this input from all active sets in-place and collect targets
// that became ready without cloning all keys first.
active_target_inputs.retain(|target_name, inputs| {
inputs.remove(&input_name);
if inputs.is_empty() {
if pending_targets.remove(target_name) {
targets_to_trigger.push(target_name.clone());
}
false
} else {
true
}
}
});
// Deduplicate: only spawn for targets that don't already have a running trigger task
let targets_to_spawn: Vec<String> = {
let mut running = running_trigger_targets.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let mut pending =
pending_trigger_targets.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
let mut to_spawn = Vec::new();
for target in targets_to_trigger {
if running.insert(target.clone()) {
if running_trigger_targets.insert(target.clone()) {
to_spawn.push(target);
} else {
pending.insert(target);
pending_trigger_targets.insert(target);
}
}
to_spawn
backend/src/api/model/active_provider_manager.rs
+26 -38
View File
@@ -9,7 +9,7 @@ use crate::{
use log::{error, warn};
use shared::utils::sanitize_sensitive_info;
use std::cmp::Reverse;
use std::collections::{BTreeMap, HashMap};
use std::collections::{BTreeMap, HashMap, HashSet};
use std::{
net::SocketAddr,
sync::{
@@ -75,18 +75,18 @@ struct SharedAllocation {
#[derive(Debug, Clone)]
struct ActiveConnectionInfo {
allocation: ProviderAllocation,
priority: i8,
is_probe: bool,
// Used to signal preemption to the consumer of this connection
cancel_token: CancellationToken,
created_at: Instant,
priority: i8,
is_probe: bool,
}
#[derive(Debug, Clone, Default)]
struct SharedConnections {
by_key: HashMap<String, SharedAllocation>,
key_by_addr: HashMap<ClientConnectionId, String>,
shared_by_allocation_id: HashMap<AllocationId, String>,
by_key: HashMap<Arc<str>, SharedAllocation>,
key_by_addr: HashMap<ClientConnectionId, Arc<str>>,
shared_by_allocation_id: HashMap<AllocationId, Arc<str>>,
}
#[derive(Debug, Clone, Default)]
@@ -95,8 +95,8 @@ struct Connections {
single: HashMap<ClientConnectionId, HashMap<AllocationId, ActiveConnectionInfo>>,
shared: SharedConnections,
// Index to quickly find connections by provider name for preemption
// ProviderName -> Vec<(ClientConnectionId, AllocationId)>
by_provider: HashMap<Arc<str>, Vec<(ClientConnectionId, AllocationId)>>,
// ProviderName -> Set<(ClientConnectionId, AllocationId)>
by_provider: HashMap<Arc<str>, HashSet<(ClientConnectionId, AllocationId)>>,
// Priority index per provider alias for O(log n) victim lookup
// ProviderName -> BTreeMap<PriorityKey, PriorityOwner>
priority_index: HashMap<Arc<str>, BTreeMap<PriorityKey, PriorityOwner>>,
@@ -248,14 +248,14 @@ impl ActiveProviderManager {
allocation_id,
ActiveConnectionInfo {
allocation: allocation.clone(),
priority,
is_probe,
cancel_token: cancel_token.clone(),
created_at: now,
priority,
is_probe,
},
);
connections.by_provider.entry(provider_name.clone()).or_default().push((*addr, allocation_id));
connections.by_provider.entry(provider_name.clone()).or_default().insert((*addr, allocation_id));
connections.priority_index.entry(provider_name.clone()).or_default()
.insert((priority, Reverse(now), allocation_id), PriorityOwner::Single(*addr));
@@ -403,9 +403,7 @@ impl ActiveProviderManager {
}
if let Some(ref name) = removed_provider_name {
if let Some(list) = connections.by_provider.get_mut(name) {
if let Some(idx) = list.iter().position(|(a, i)| *a == addr && *i == alloc_id) {
list.remove(idx);
}
list.remove(&(addr, alloc_id));
}
if let Some(tree) = connections.priority_index.get_mut(name) {
tree.remove(&(v_prio, Reverse(victim_created_at), alloc_id));
@@ -577,9 +575,7 @@ impl ActiveProviderManager {
}
if let Some(name) = removed_provider_name {
if let Some(list) = connections.by_provider.get_mut(&name) {
if let Some(idx) = list.iter().position(|(a, i)| *a == addr && *i == alloc_id) {
list.remove(idx);
}
list.remove(&(addr, alloc_id));
}
if let Some(tree) = connections.priority_index.get_mut(&name) {
tree.remove(&(v_prio, Reverse(victim_created_at), alloc_id));
@@ -700,9 +696,7 @@ impl ActiveProviderManager {
for (id, info) in &allocations {
if let Some(name) = info.allocation.get_provider_name() {
if let Some(list) = connections.by_provider.get_mut(&name) {
if let Some(idx) = list.iter().position(|(a, i)| *a == *addr && *i == *id) {
list.remove(idx);
}
list.remove(&(*addr, *id));
}
if let Some(tree) = connections.priority_index.get_mut(&name) {
tree.remove(&(info.priority, Reverse(info.created_at), *id));
@@ -812,11 +806,7 @@ impl ActiveProviderManager {
// Remove from by_provider index
if let Some(name) = released.as_ref().and_then(ProviderAllocation::get_provider_name) {
if let Some(list) = connections.by_provider.get_mut(&name) {
if let Some(idx) =
list.iter().position(|(a, i)| *a == handle.client_id && *i == handle.allocation_id)
{
list.remove(idx);
}
list.remove(&(handle.client_id, handle.allocation_id));
}
released_priority_key = Some((name, pkey));
}
@@ -859,6 +849,7 @@ impl ActiveProviderManager {
let extras = {
let mut connections = self.connections.write().await;
let mut extras = Vec::new();
let shared_key: Arc<str> = Arc::from(key);
// Find the allocation to promote (must be single)
// Logic change: we must find the specific allocation if multiple exist, but usually per client only 1 active?
@@ -875,9 +866,7 @@ impl ActiveProviderManager {
// Cleanup indices
if let Some(name) = info.allocation.get_provider_name() {
if let Some(list) = connections.by_provider.get_mut(&name) {
if let Some(idx) = list.iter().position(|(a, alloc_id)| *a == *addr && *alloc_id == id) {
list.remove(idx);
}
list.remove(&(*addr, id));
}
// Remove old single entry from priority_index
if let Some(tree) = connections.priority_index.get_mut(&name) {
@@ -888,12 +877,7 @@ impl ActiveProviderManager {
for (extra_id, extra_info) in &extra_entries {
if let Some(extra_provider_name) = extra_info.allocation.get_provider_name() {
if let Some(list) = connections.by_provider.get_mut(&extra_provider_name) {
if let Some(idx) = list
.iter()
.position(|(extra_addr, alloc_id)| *extra_addr == *addr && *alloc_id == *extra_id)
{
list.remove(idx);
}
list.remove(&(*addr, *extra_id));
}
if let Some(tree) = connections.priority_index.get_mut(&extra_provider_name) {
tree.remove(&(extra_info.priority, Reverse(extra_info.created_at), *extra_id));
@@ -902,6 +886,7 @@ impl ActiveProviderManager {
}
for (_, extra_info) in extra_entries {
extra_info.cancel_token.cancel();
extras.push(extra_info.allocation);
}
@@ -929,7 +914,7 @@ impl ActiveProviderManager {
);
connections.shared.by_key.insert(
key.to_string(),
Arc::clone(&shared_key),
SharedAllocation {
allocation_id: handle.0.allocation_id,
allocation: handle.0.allocation.clone(),
@@ -939,8 +924,11 @@ impl ActiveProviderManager {
cancel_token: handle.0.cancel_token.clone(),
},
);
connections.shared.key_by_addr.insert(*addr, key.to_string());
connections.shared.shared_by_allocation_id.insert(handle.0.allocation_id, key.to_string());
connections.shared.key_by_addr.insert(*addr, Arc::clone(&shared_key));
connections
.shared
.shared_by_allocation_id
.insert(handle.0.allocation_id, shared_key);
// Insert new shared entry into priority_index
connections.priority_index.entry(provider_name.clone()).or_default()
@@ -1004,7 +992,7 @@ impl ActiveProviderManager {
}
}
connections.shared.key_by_addr.insert(*addr, key.to_string());
connections.shared.key_by_addr.insert(*addr, Arc::from(key));
Ok(())
}
backend/src/api/model/connection_manager.rs
+26 -13
View File
@@ -14,6 +14,7 @@ use shared::{
use std::{borrow::Cow, net::SocketAddr, sync::Arc};
use tokio::sync::{mpsc, Notify};
const CLEANUP_QUEUE_CAPACITY: usize = 4096;
fn notify_capacity(capacity_notify: &Notify) { capacity_notify.notify_waiters(); }
pub(crate) enum CleanupEvent {
@@ -37,7 +38,7 @@ pub struct ConnectionManager {
pub shared_stream_manager: Arc<SharedStreamManager>,
event_manager: Arc<EventManager>,
close_socket_signal_tx: tokio::sync::broadcast::Sender<SocketAddr>,
cleanup_tx: mpsc::UnboundedSender<CleanupEvent>,
cleanup_tx: mpsc::Sender<CleanupEvent>,
capacity_notify: Arc<Notify>,
}
@@ -49,7 +50,7 @@ impl ConnectionManager {
event_manager: &Arc<EventManager>,
) -> Self {
let (close_socket_signal_tx, _) = tokio::sync::broadcast::channel(256);
let (cleanup_tx, cleanup_rx) = mpsc::unbounded_channel();
let (cleanup_tx, cleanup_rx) = mpsc::channel(CLEANUP_QUEUE_CAPACITY);
let capacity_notify = Arc::new(Notify::new());
let mgr = Self {
@@ -75,7 +76,7 @@ impl ConnectionManager {
}
fn spawn_cleanup_worker(
mut rx: mpsc::UnboundedReceiver<CleanupEvent>,
mut rx: mpsc::Receiver<CleanupEvent>,
user_manager: Arc<ActiveUserManager>,
provider_manager: Arc<ActiveProviderManager>,
shared_stream_manager: Arc<SharedStreamManager>,
@@ -103,15 +104,21 @@ impl ConnectionManager {
// Release provider handle first to avoid a race window where the user
// connection count drops (making capacity appear available) before the
// provider slot is actually freed.
if let Some(h) = handle {
let provider_released = if let Some(h) = handle {
provider_manager.release_handle(&h).await;
}
if user_manager.release_stream(&addr).await {
true
} else {
false
};
let stream_released = user_manager.release_stream(&addr).await;
if stream_released {
event_manager.send_event(EventMessage::ActiveUser(
ActiveUserConnectionChange::Disconnected(addr),
));
}
notify_capacity(capacity_notify.as_ref());
if provider_released || stream_released {
notify_capacity(capacity_notify.as_ref());
}
}
CleanupEvent::UpdateDetailAndReleaseProvider { addr, video_type, handle } => {
if let Some(stream_info) = user_manager.update_stream_detail(&addr, video_type).await {
@@ -141,8 +148,14 @@ impl ConnectionManager {
}
pub(crate) fn send_cleanup(&self, event: CleanupEvent) {
if let Err(e) = self.cleanup_tx.send(event) {
debug!("Cleanup channel closed, dropping event: {e}");
match self.cleanup_tx.try_send(event) {
Ok(()) => {}
Err(tokio::sync::mpsc::error::TrySendError::Full(_event)) => {
warn!("Cleanup queue full, dropping event");
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(_event)) => {
debug!("Cleanup channel closed, dropping cleanup event");
}
}
}
@@ -176,26 +189,26 @@ impl ConnectionManager {
if removed {
self.event_manager.send_event(EventMessage::ActiveUser(ActiveUserConnectionChange::Disconnected(*addr)));
}
self.capacity_notify.notify_waiters();
notify_capacity(self.capacity_notify.as_ref());
}
pub async fn release_provider_connection(&self, addr: &SocketAddr) {
self.provider_manager.release_connection(addr).await;
self.shared_stream_manager.release_connection(addr, false).await;
self.capacity_notify.notify_waiters();
notify_capacity(self.capacity_notify.as_ref());
}
pub async fn release_stream(&self, addr: &SocketAddr) {
if self.user_manager.release_stream(addr).await {
self.event_manager.send_event(EventMessage::ActiveUser(ActiveUserConnectionChange::Disconnected(*addr)));
self.capacity_notify.notify_waiters();
notify_capacity(self.capacity_notify.as_ref());
}
}
pub async fn release_provider_handle(&self, provider_handle: Option<ProviderHandle>) {
if let Some(handle) = provider_handle {
self.provider_manager.release_handle(&handle).await;
self.capacity_notify.notify_waiters();
notify_capacity(self.capacity_notify.as_ref());
}
}
backend/src/api/model/stream_error.rs
+1 -1
View File
@@ -29,7 +29,7 @@ impl std::fmt::Display for StreamError {
StreamError::StdIo(e) => write!(f, "IO error: {e}"),
// StreamError::ReceiverClosed => write!(f, "Receiver closed"),
StreamError::ReceiverError(e) => write!(f, "Receiver error {e}"),
StreamError::Stream(e) => write!(f, "Stream error: {e}"),
StreamError::Stream(e) => write!(f, "Stream: {e}"),
StreamError::LockError(e) => write!(f, "LockError: {e}"),
StreamError::MalformedPacket(e) => write!(f, "MalformedPacket: {e}"),
StreamError::InvalidTimestamp(e) => write!(f, "InvalidTimestamp: {e}"),
backend/src/api/model/streams/active_client_stream.rs
+19 -7
View File
@@ -157,7 +157,6 @@ impl ActiveClientStreamState {
self.provider_stopped = true;
self.preempt_cancelled = None;
self.stop_grace_task();
self.release_user_stream();
let mut serve_preempted_custom = false;
if self.provider_handle.is_some() {
@@ -188,11 +187,16 @@ impl ActiveClientStreamState {
"Provider stream preempted for {}; stopping client stream",
sanitize_sensitive_info(&addr.to_string())
);
self.connection_manager.send_cleanup(CleanupEvent::UpdateDetailAndReleaseProvider {
addr,
video_type: CustomVideoStreamType::LowPriorityPreempted,
handle,
});
if serve_preempted_custom {
self.connection_manager.send_cleanup(CleanupEvent::UpdateDetailAndReleaseProvider {
addr,
video_type: CustomVideoStreamType::LowPriorityPreempted,
handle,
});
} else {
self.release_user_stream();
self.connection_manager.send_cleanup(CleanupEvent::ReleaseProviderHandle { handle });
}
}
serve_preempted_custom
}
@@ -577,12 +581,20 @@ pub(crate) async fn create_active_client_stream(
.and_then(|h| h.cancel_token.as_ref())
.map(|token| Box::pin(token.clone().cancelled_owned()));
let mut send_custom_stream_flag = grace_stop_flag;
if send_custom_stream_flag.is_none()
&& preempt_cancelled.is_some()
&& custom_video.low_priority_preempted.is_some()
{
send_custom_stream_flag = Some(Arc::new(AtomicU8::new(StreamMode::Inner as u8)));
}
let state = ActiveClientStreamState {
inner: stream,
preempt_cancelled,
grace_task_handle,
provider_handle: stream_details.provider_handle,
send_custom_stream_flag: grace_stop_flag,
send_custom_stream_flag,
provisionable: has_provisioning,
custom_video,
waker: Some(waker),
backend/src/api/model/streams/shared_stream_manager.rs
+138 -28
View File
@@ -69,7 +69,7 @@ fn convert_stream(stream: BoxStream<Bytes>) -> BoxStream<Result<Bytes, StreamErr
type SubscriberId = SocketAddr;
struct BurstBuffer {
buffer: VecDeque<Arc<Bytes>>,
buffer: VecDeque<Bytes>,
buffer_size: usize,
current_bytes: usize,
}
@@ -89,9 +89,9 @@ impl BurstBuffer {
Self { buffer: VecDeque::with_capacity(buf_size), buffer_size: buf_size, current_bytes: 0 }
}
pub fn snapshot(&self) -> VecDeque<Arc<Bytes>> { self.buffer.iter().cloned().collect::<VecDeque<Arc<Bytes>>>() }
pub fn snapshot(&self) -> VecDeque<Bytes> { self.buffer.iter().cloned().collect::<VecDeque<Bytes>>() }
pub fn push(&mut self, packet: Arc<Bytes>) {
pub fn push(&mut self, packet: Bytes) {
while self.current_bytes + packet.len() > self.buffer_size {
if let Some(popped) = self.buffer.pop_front() {
self.current_bytes -= popped.len();
@@ -106,7 +106,7 @@ impl BurstBuffer {
}
async fn send_burst_buffer(
start_buffer: &VecDeque<Arc<Bytes>>,
start_buffer: &VecDeque<Bytes>,
client_tx: &Sender<Bytes>,
cancellation_token: &CancellationToken,
) -> usize {
@@ -115,7 +115,7 @@ async fn send_burst_buffer(
if cancellation_token.is_cancelled() {
return sent;
}
if let Err(err) = client_tx.send(buf.as_ref().clone()).await {
if let Err(err) = client_tx.send(buf.clone()).await {
debug!("Failed sending burst-buffer chunk to client: {err}");
return sent; // stop on send error
}
@@ -132,6 +132,8 @@ pub struct SharedStreamState {
headers: Vec<(String, String)>,
buf_size: usize,
provider_guard: Option<ProviderHandle>,
low_priority_preempted: Option<crate::api::model::TransportStreamBuffer>,
preempted_token: CancellationToken,
subscribers: RwLock<HashMap<SubscriberId, CancellationToken>>,
broadcaster: tokio::sync::broadcast::Sender<Bytes>,
stop_token: CancellationToken,
@@ -145,6 +147,7 @@ impl SharedStreamState {
buf_size: usize,
provider_guard: Option<ProviderHandle>,
min_burst_buffer_size: usize,
low_priority_preempted: Option<crate::api::model::TransportStreamBuffer>,
) -> Self {
let (broadcaster, _) = tokio::sync::broadcast::channel(buf_size);
// TODO channel size versus byte size, channels are chunk sized, burst_buffer byte sized
@@ -153,6 +156,8 @@ impl SharedStreamState {
headers,
buf_size,
provider_guard,
low_priority_preempted,
preempted_token: CancellationToken::new(),
subscribers: RwLock::new(HashMap::new()),
broadcaster,
stop_token: CancellationToken::new(),
@@ -197,6 +202,8 @@ impl SharedStreamState {
let mut last_lag_log = Instant::now().checked_sub(Duration::from_secs(10)).unwrap_or_else(Instant::now);
let mut consecutive_lag_count: u32 = 0;
let subscriber_buf_size = self.buf_size;
let preempted_token = self.preempted_token.clone();
let low_priority_preempted = self.low_priority_preempted.clone();
let address = *addr;
let subscriber_started_at = Instant::now();
@@ -310,7 +317,35 @@ impl SharedStreamState {
.min(1_600);
sleep(Duration::from_millis(backoff_ms)).await;
}
Err(_) => break,
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
if preempted_token.is_cancelled() {
if let Some(mut fallback) = low_priority_preempted {
debug_if_enabled!(
"Shared stream subscriber {} switching to low_priority_preempted fallback",
sanitize_sensitive_info(&address.to_string())
);
loop {
if cancel_token.is_cancelled() || client_tx_clone.is_closed() {
break;
}
if let Some(chunk) = fallback.next_chunk() {
if let Err(err) = client_tx.send(chunk).await {
debug!(
"Shared stream fallback send error for {}: {err}",
sanitize_sensitive_info(&address.to_string())
);
break;
}
} else {
// End of fallback stream; close subscriber task.
break;
}
}
}
}
break;
}
}
}
}
@@ -392,13 +427,12 @@ impl SharedStreamState {
startup_stats_logged = true;
}
}
let arc_data = Arc::new(data);
{
let mut buffer = burst_buffer.write().await;
buffer.push(arc_data.clone());
buffer.push(data.clone());
}
match sender.send(arc_data.as_ref().clone()) {
match sender.send(data) {
Ok(clients) => {
if clients == 0 {
debug_if_enabled!("No shared stream subscribers closing {}", sanitize_sensitive_info(&streaming_url));
@@ -446,8 +480,8 @@ impl SharedStreamState {
#[derive(Debug, Clone, Default)]
struct SharedStreamsRegister {
by_key: HashMap<String, Arc<SharedStreamState>>,
key_by_addr: HashMap<SubscriberId, String>,
by_key: HashMap<Arc<str>, Arc<SharedStreamState>>,
key_by_addr: HashMap<SubscriberId, Arc<str>>,
}
pub struct SharedStreamManager {
@@ -475,7 +509,7 @@ impl SharedStreamManager {
let remove_keys: Vec<SocketAddr> = shared_streams
.key_by_addr
.iter()
.filter_map(|(addr, url)| if url == stream_url { Some(*addr) } else { None })
.filter_map(|(addr, url)| if url.as_ref() == stream_url { Some(*addr) } else { None })
.collect();
for k in remove_keys {
shared_streams.key_by_addr.remove(&k);
@@ -524,7 +558,7 @@ impl SharedStreamManager {
let remove_keys: Vec<SocketAddr> = shared_streams
.key_by_addr
.iter()
.filter_map(|(addr, url)| if url == stream_url { Some(*addr) } else { None })
.filter_map(|(addr, url)| if url.as_ref() == stream_url { Some(*addr) } else { None })
.collect();
for k in remove_keys {
shared_streams.key_by_addr.remove(&k);
@@ -539,9 +573,7 @@ impl SharedStreamManager {
sanitize_sensitive_info(stream_url)
);
for handle in shared_state.task_handles.write().await.drain(..) {
handle.abort();
}
shared_state.preempted_token.cancel();
// Cancel stop_token to terminate the broadcast task.
// Do NOT release provider_guard — the preemption caller already released the allocation.
@@ -610,8 +642,12 @@ impl SharedStreamManager {
) -> Option<(BoxedProviderStream, Option<Arc<str>>)> {
let shared_state_opt = {
let mut shared_streams = self.shared_streams.write().await;
if let Some(shared_state) = shared_streams.by_key.get(stream_url).cloned() {
shared_streams.key_by_addr.insert(*addr, stream_url.to_owned());
if let Some((stream_key, shared_state)) = shared_streams
.by_key
.get_key_value(stream_url)
.map(|(stream_key, shared_state)| (Arc::clone(stream_key), Arc::clone(shared_state)))
{
shared_streams.key_by_addr.insert(*addr, stream_key);
Some(shared_state)
} else {
None
@@ -632,8 +668,9 @@ impl SharedStreamManager {
async fn register(&self, addr: &SocketAddr, stream_url: &str, shared_state: Arc<SharedStreamState>) {
let mut shared_streams = self.shared_streams.write().await;
shared_streams.by_key.insert(stream_url.to_string(), shared_state);
shared_streams.key_by_addr.insert(*addr, stream_url.to_string());
let stream_key: Arc<str> = Arc::from(stream_url);
shared_streams.by_key.insert(Arc::clone(&stream_key), shared_state);
shared_streams.key_by_addr.insert(*addr, stream_key);
debug_if_enabled!(
"Registered shared stream {} for initial subscriber {}",
sanitize_sensitive_info(stream_url),
@@ -660,7 +697,19 @@ impl SharedStreamManager {
let buf_size = CHANNEL_SIZE.max(buffer_size);
let config = app_state.app_config.config.load();
let min_buffer_bytes = resolve_min_burst_buffer_bytes(&config);
let shared_state = Arc::new(SharedStreamState::new(headers, buf_size, provider_handle, min_buffer_bytes));
let low_priority_preempted = app_state
.app_config
.custom_stream_response
.load()
.as_ref()
.and_then(|c| c.low_priority_preempted.clone());
let shared_state = Arc::new(SharedStreamState::new(
headers,
buf_size,
provider_handle,
min_buffer_bytes,
low_priority_preempted,
));
app_state.shared_stream_manager.register(addr, stream_url, Arc::clone(&shared_state)).await;
app_state.active_provider.make_shared_connection(addr, stream_url).await;
let subscribed_stream = Self::subscribe_shared_stream(app_state, stream_url, addr, user_priority).await;
@@ -675,6 +724,11 @@ impl SharedStreamManager {
"Created shared provider stream {} (channel_capacity={buf_size}, burst_buffer_min={min_buffer_bytes} bytes)",
sanitize_sensitive_info(stream_url)
);
} else {
warn!(
"Shared stream subscribe failed for {}; broadcaster will not start",
sanitize_sensitive_info(stream_url)
);
}
subscribed_stream
}
@@ -716,11 +770,13 @@ mod tests {
utils::FileLockManager,
};
use arc_swap::{ArcSwap, ArcSwapOption};
use futures::StreamExt;
use shared::{
model::{ConfigPaths, InputFetchMethod, InputType},
utils::Internable,
};
use std::{collections::HashMap, net::SocketAddr, sync::Arc};
use tokio::time::{timeout, Duration};
use tokio_util::sync::CancellationToken;
fn create_test_app_config() -> AppConfig {
@@ -779,13 +835,19 @@ mod tests {
let addr_1: SocketAddr = "127.0.0.1:41001".parse().unwrap_or_else(|_| unreachable!());
let addr_2: SocketAddr = "127.0.0.1:41002".parse().unwrap_or_else(|_| unreachable!());
let state = Arc::new(SharedStreamState::new(Vec::new(), CHANNEL_SIZE.max(8), None, 1024));
let state = Arc::new(SharedStreamState::new(
Vec::new(),
CHANNEL_SIZE.max(8),
None,
1024,
None,
));
{
let mut reg = shared_manager.shared_streams.write().await;
reg.by_key.insert(stream_url.to_string(), Arc::clone(&state));
reg.key_by_addr.insert(addr_1, stream_url.to_string());
reg.key_by_addr.insert(addr_2, stream_url.to_string());
reg.by_key.insert(Arc::from(stream_url), Arc::clone(&state));
reg.key_by_addr.insert(addr_1, Arc::from(stream_url));
reg.key_by_addr.insert(addr_2, Arc::from(stream_url));
}
{
@@ -822,12 +884,18 @@ mod tests {
let stream_url = "https://example.invalid/live/single.ts";
let addr_1: SocketAddr = "127.0.0.1:42001".parse().unwrap_or_else(|_| unreachable!());
let state = Arc::new(SharedStreamState::new(Vec::new(), CHANNEL_SIZE.max(8), None, 1024));
let state = Arc::new(SharedStreamState::new(
Vec::new(),
CHANNEL_SIZE.max(8),
None,
1024,
None,
));
{
let mut reg = shared_manager.shared_streams.write().await;
reg.by_key.insert(stream_url.to_string(), Arc::clone(&state));
reg.key_by_addr.insert(addr_1, stream_url.to_string());
reg.by_key.insert(Arc::from(stream_url), Arc::clone(&state));
reg.key_by_addr.insert(addr_1, Arc::from(stream_url));
}
{
@@ -857,4 +925,46 @@ mod tests {
assert!(subs.is_empty(), "duplicate release must keep subscribers empty");
}
}
#[tokio::test]
async fn test_preempted_shared_subscriber_switches_to_low_priority_fallback() {
let app_cfg = create_test_app_config();
let event_manager = Arc::new(EventManager::new());
let provider_manager = Arc::new(ActiveProviderManager::new(&app_cfg, &event_manager));
let shared_manager = Arc::new(SharedStreamManager::new(provider_manager));
let addr: SocketAddr = "127.0.0.1:43001".parse().unwrap_or_else(|_| unreachable!());
// Minimal valid TS packet (sync byte + padding) to feed TransportStreamBuffer.
let mut ts_packet = vec![0_u8; 188];
ts_packet[0] = 0x47;
let low_priority_fallback = crate::api::model::TransportStreamBuffer::new(ts_packet);
let state = Arc::new(SharedStreamState::new(
Vec::new(),
CHANNEL_SIZE.max(8),
None,
1024,
Some(low_priority_fallback),
));
let (mut stream, _provider) = state.subscribe(&addr, Arc::clone(&shared_manager)).await;
// Simulate preemption teardown semantics:
// 1) mark as preempted, 2) drop state so broadcaster sender closes.
state.preempted_token.cancel();
drop(state);
let first = timeout(Duration::from_secs(2), stream.next()).await;
let maybe_chunk = match first {
Ok(v) => v,
Err(_) => panic!("timed out waiting for fallback chunk"),
};
let chunk = match maybe_chunk {
Some(Ok(bytes)) => bytes,
Some(Err(err)) => panic!("fallback stream returned error: {err}"),
None => panic!("fallback stream ended unexpectedly"),
};
assert!(!chunk.is_empty(), "fallback chunk must contain MPEG-TS bytes");
}
}
backend/src/api/model/streams/transport_stream_buffer.rs
+79 -19
View File
@@ -288,6 +288,46 @@ pub fn calculate_duration_ticks(buffer: &[u8], packet_indices: &PacketIndices) -
}
}
fn calculate_pcr_duration_ticks(buffer: &[u8]) -> Option<u64> {
let mut first_pcr = None;
let mut last_pcr = None;
let mut i = 0;
while i + TS_PACKET_SIZE <= buffer.len() {
if buffer[i] != SYNC_BYTE {
i += 1;
continue;
}
let packet = &buffer[i..i + TS_PACKET_SIZE];
let afc = (packet[3] >> 4) & 0b11;
if afc == 2 || afc == 3 {
let adaptation_len = packet[4] as usize;
if adaptation_len >= 7 && (packet[5] & ADAPTATION_FIELD_FLAG_PCR) != 0 {
let pcr = decode_pcr(&packet[6..12]);
if first_pcr.is_none() {
first_pcr = Some(pcr);
}
last_pcr = Some(pcr);
}
}
i += TS_PACKET_SIZE;
}
let (Some(first), Some(last)) = (first_pcr, last_pcr) else {
return None;
};
if first == last {
return None;
}
let diff_27mhz = if last >= first {
last - first
} else {
MAX_PCR - first + last
};
let ticks_90khz = diff_27mhz / 300;
(ticks_90khz > 0).then_some(ticks_90khz)
}
type PacketIndices = Vec<(usize, Option<(usize, Option<usize>, u16)>)>;
pub struct TransportStreamBuffer {
@@ -303,6 +343,7 @@ pub struct TransportStreamBuffer {
cc_entries: Box<[Option<(u8, bool)>; 8192]>,
waker: Arc<AtomicWaker>,
first_pcr: Option<u64>,
force_discontinuity_on_wrap: bool,
}
impl std::fmt::Debug for TransportStreamBuffer {
@@ -314,6 +355,7 @@ impl std::fmt::Debug for TransportStreamBuffer {
.field("timestamp_offset", &self.timestamp_offset)
.field("stream_duration_90khz", &self.stream_duration_90khz)
.field("first_pcr", &self.first_pcr)
.field("force_discontinuity_on_wrap", &self.force_discontinuity_on_wrap)
.finish_non_exhaustive()
}
}
@@ -333,6 +375,9 @@ impl Clone for TransportStreamBuffer {
cc_entries: Box::new([None; 8192]),
waker: Arc::clone(&self.waker),
first_pcr: self.first_pcr,
// Start each clone with one initial discontinuity marker to keep
// continuity behavior consistent for fresh consumers.
force_discontinuity_on_wrap: true,
}
}
}
@@ -349,7 +394,10 @@ impl TransportStreamBuffer {
let (packet_indices, _) = extract_pts_dts_indices_with_continuity(&raw);
let length = packet_indices.len();
let stream_duration_90khz = calculate_duration_ticks(&raw, &packet_indices);
let mut stream_duration_90khz = calculate_duration_ticks(&raw, &packet_indices);
if stream_duration_90khz == 0 {
stream_duration_90khz = calculate_pcr_duration_ticks(&raw).unwrap_or(0);
}
// Scan for the first PCR in the buffer to use as a reference for discontinuity packets
let mut first_pcr = None;
@@ -383,6 +431,9 @@ impl TransportStreamBuffer {
cc_entries: Box::new([None; 8192]),
waker: Arc::new(AtomicWaker::new()),
first_pcr,
// Emit a discontinuity packet on startup. Subsequent injections are
// governed by wrap handling and duration/discontinuity logic.
force_discontinuity_on_wrap: true,
}
}
@@ -549,9 +600,19 @@ impl TransportStreamBuffer {
// increasing across loops. Resetting to 0 causes decoders (MPV, ffmpeg) to see
// a backward timestamp jump and treat the loop as end-of-stream or corrupt data.
self.current_pos = 0;
self.timestamp_offset =
self.timestamp_offset.wrapping_add(self.stream_duration_90khz) % MAX_PTS_DTS;
self.current_dts = 0;
// Advance timestamps by one full source duration per loop so output time is
// monotonic for clients. Resetting to zero causes backward jumps that some
// players interpret as stream end/corruption after the first cycle.
if self.stream_duration_90khz > 0 {
self.timestamp_offset =
self.timestamp_offset.wrapping_add(self.stream_duration_90khz) % MAX_PTS_DTS;
self.current_dts = self.current_dts.wrapping_add(self.stream_duration_90khz) % MAX_PTS_DTS;
} else {
// PCR-only (or malformed) assets may not expose PTS/DTS-derived duration.
// Force one discontinuity marker after wrap so decoders do not see identical
// timestamp cycles as a continuous timeline.
self.force_discontinuity_on_wrap = true;
}
// Reset only the discontinuity-sent flag so injection packets are emitted at the
// start of the next loop. Continuity counter values keep running so CC remains
@@ -577,24 +638,23 @@ impl TransportStreamBuffer {
}
let (counter, discontinuity_sent) = entry.as_mut().unwrap();
// Inject a discontinuity packet for every PID on first appearance and at loop
// boundaries. This covers both PES PIDs (audio/video) and PSI PIDs (PAT/PMT)
// so that decoders accept the CC reset during mid-stream switches.
let inject_discontinuity = !*discontinuity_sent;
let payload_packet_cc;
// Disable synthetic discontinuity packet insertion for looped custom streams.
// In practice this can produce demuxer corruption on some clients (PES mismatch).
// Monotonic timestamps + stable continuity counters are sufficient here.
let inject_discontinuity = self.force_discontinuity_on_wrap && !*discontinuity_sent;
if inject_discontinuity {
// Discontinuity packet is adaptation-only and does not consume a CC step.
// Keep CC progression tied strictly to payload packets.
let extra_packet_cc = if packet_has_payload { *counter } else { packet[3] & 0x0F };
payload_packet_cc = if packet_has_payload { *counter } else { packet[3] & 0x0F };
*discontinuity_sent = true;
// Write discontinuity packet directly into the output buffer — no allocation.
Self::generate_discontinuity_packet(packet, extra_packet_cc, self.first_pcr, self.timestamp_offset, &mut bytes);
} else {
payload_packet_cc = if packet_has_payload { *counter } else { packet[3] & 0x0F };
Self::generate_discontinuity_packet(
packet,
extra_packet_cc,
self.first_pcr,
self.timestamp_offset,
&mut bytes,
);
self.force_discontinuity_on_wrap = false;
}
*discontinuity_sent = true;
let payload_packet_cc = if packet_has_payload { *counter } else { packet[3] & 0x0F };
// TS continuity counter increments only when payload is present (AFC=01/11).
if packet_has_payload {
backend/src/model/config/app.rs
+4 -1
View File
@@ -12,6 +12,10 @@ use rand::Rng;
use shared::error::{TuliproxError, TuliproxErrorKind};
use shared::info_err_res;
use shared::model::ConfigPaths;
use shared::utils::{
CHANNEL_UNAVAILABLE, LOW_PRIORITY_PREEMPTED, PANEL_API_PROVISIONING, PROVIDER_CONNECTIONS_EXHAUSTED,
USER_ACCOUNT_EXPIRED, USER_CONNECTIONS_EXHAUSTED,
};
use std::borrow::Cow;
use std::collections::HashSet;
use std::fs::File;
@@ -19,7 +23,6 @@ use std::io::Read;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::sync::OnceCell;
use shared::utils::{CHANNEL_UNAVAILABLE, LOW_PRIORITY_PREEMPTED, PANEL_API_PROVISIONING, PROVIDER_CONNECTIONS_EXHAUSTED, USER_ACCOUNT_EXPIRED, USER_CONNECTIONS_EXHAUSTED};
fn generate_secret() -> [u8; 32] {
let mut rng = rand::rng();
backend/src/utils/network/request.rs
+35 -27
View File
@@ -1611,32 +1611,6 @@ mod tests {
Arc::new(ConfigProvider::from(&dto))
}
async fn start_plain_http_server() -> (SocketAddr, Arc<AtomicUsize>, tokio::task::JoinHandle<()>) {
let listener = TcpListener::bind("127.0.0.1:0").await.expect("tcp bind should work");
let addr = listener.local_addr().expect("local addr should exist");
let accepted = Arc::new(AtomicUsize::new(0));
let accepted_clone = Arc::clone(&accepted);
let handle = tokio::spawn(async move {
loop {
let Ok((mut socket, _)) = listener.accept().await else {
continue;
};
accepted_clone.fetch_add(1, Ordering::SeqCst);
tokio::spawn(async move {
let mut buf = vec![0_u8; 2048];
let _ = socket.read(&mut buf).await;
let _ = socket
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
.await;
let _ = socket.shutdown().await;
});
}
});
(addr, accepted, handle)
}
#[test]
fn test_url_mask() {
// Replace with "***"
@@ -1816,9 +1790,43 @@ mod tests {
assert_eq!(https_second.connect_ip.map(|ip| ip.to_string()), Some("203.0.113.11".to_string()));
}
async fn start_plain_http_server() -> std::io::Result<(SocketAddr, Arc<AtomicUsize>, tokio::task::JoinHandle<()>)> {
let listener = TcpListener::bind("127.0.0.1:0").await?;
let addr = listener.local_addr()?;
let accepted = Arc::new(AtomicUsize::new(0));
let accepted_clone = Arc::clone(&accepted);
let handle = tokio::spawn(async move {
loop {
let Ok((mut socket, _)) = listener.accept().await else {
continue;
};
accepted_clone.fetch_add(1, Ordering::SeqCst);
tokio::spawn(async move {
let mut buf = vec![0_u8; 2048];
let _ = socket.read(&mut buf).await;
let _ = socket
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok")
.await;
let _ = socket.shutdown().await;
});
}
});
Ok((addr, accepted, handle))
}
#[tokio::test]
async fn test_on_connect_error_try_next_ip_before_provider_rotation() {
let (addr, accepted, server_handle) = start_plain_http_server().await;
let (addr, accepted, server_handle) = match start_plain_http_server().await {
Ok(server) => server,
Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => {
eprintln!("skipping test_on_connect_error_try_next_ip_before_provider_rotation: {err}");
return;
}
Err(err) => panic!("failed to start test http server: {err}"),
};
let mut cfg = Config {
connect_timeout_secs: 1,
bin/build_resources.sh
+2 -2
View File
@@ -22,7 +22,7 @@ while getopts "fh" opt; do
esac
done
declare -a resources=("channel_unavailable" "user_connections_exhausted" "provider_connections_exhausted" "user_account_expired" "panel_api_provisioning" "low_priority_preempted")
declare -a resources=("channel_unavailable" "user_connections_exhausted" "provider_connections_exhausted" "low_priority_preempted" "user_account_expired" "panel_api_provisioning")
for resource in "${resources[@]}"; do
if [ "$flag_force" = false ]; then
@@ -46,7 +46,7 @@ for resource in "${resources[@]}"; do
exit 1
fi
else
echo "ffmpeg not found"
echo "ffmpeg not found" >&2
exit 1
fi
done
bin/inc_version.sh
+45 -13
View File
@@ -1,20 +1,25 @@
#!/usr/bin/env bash
set -eo pipefail
set -euo pipefail
WORKING_DIR=$(pwd)
BACKEND_DIR="${WORKING_DIR}/backend"
if ! command -v cargo-set-version &> /dev/null
then
echo "🧨 cargo-set-version could not be found. Install it with 'cargo install cargo-edit'"
exit 1
fi
ROOT_TOML="${WORKING_DIR}/Cargo.toml"
# Read current version from Cargo.toml
OLD_VERSION="$(grep '^version' "${BACKEND_DIR}/Cargo.toml" | head -n1 | cut -d'"' -f2 || true)"
# Read [workspace.package].version from Cargo.toml
OLD_VERSION="$(
awk '
/^\[workspace\.package\][[:space:]]*$/ { in_ws_pkg=1; next }
/^\[[^]]+\][[:space:]]*$/ { if (in_ws_pkg) exit }
in_ws_pkg && /^[[:space:]]*version[[:space:]]*=/ {
if (match($0, /"[^"]+"/)) {
print substr($0, RSTART + 1, RLENGTH - 2)
exit
}
}
' "${ROOT_TOML}" || true
)"
if [ -z "${OLD_VERSION}" ]; then
echo "🧨 Failed to read version from '${BACKEND_DIR}/Cargo.toml' (expected a line like: version = \"x.y.z\")."
echo "🧨 Failed to read [workspace.package].version from '${ROOT_TOML}'."
exit 1
fi
@@ -42,8 +47,35 @@ esac
NEW_VERSION="${major}.${minor}.${patch}"
cargo set-version "$NEW_VERSION"
# Update only [workspace.package].version in Cargo.toml
TMP_FILE="$(mktemp)"
if awk -v new_version="${NEW_VERSION}" '
BEGIN { in_ws_pkg=0; version_updated=0 }
/^\[workspace\.package\][[:space:]]*$/ { in_ws_pkg=1; print; next }
/^\[[^]]+\][[:space:]]*$/ {
if (in_ws_pkg && !version_updated) {
in_ws_pkg=0
}
}
{
if (in_ws_pkg && !version_updated && /^[[:space:]]*version[[:space:]]*=/) {
sub(/[[:space:]]*=[[:space:]]*"[^"]*"/, " = \"" new_version "\"")
version_updated=1
}
print
}
END {
if (!version_updated) {
exit 2
}
}
' "${ROOT_TOML}" > "${TMP_FILE}"; then
mv "${TMP_FILE}" "${ROOT_TOML}"
else
rm -f "${TMP_FILE}"
echo "🧨 Failed to update [workspace.package].version in '${ROOT_TOML}'."
exit 1
fi
VERSION=v$NEW_VERSION
echo "🛠️ Set version $VERSION"