gru/pyTorrent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

surge refill

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2026-05-31 22:54:19 +02:00
parent 0477754249
commit 63c2a8f3ba
9 changed files with 291 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pytorrent/services/smart_queue.py
+251 -7
View File
@@ -154,6 +154,10 @@ def _default_settings(profile_id: int) -> dict[str, Any]:
'refill_enabled': 1,
'refill_interval_minutes': 0,
'last_refill_at': None,
'surge_refill_enabled': 0,
'surge_refill_interval_minutes': 1440,
'surge_refill_batch_size': 2000,
'last_surge_refill_at': None,
'stop_batch_size': 50,
'start_grace_seconds': 900,
'protect_active_below_cap': 1,
@@ -213,11 +217,15 @@ def save_settings(profile_id: int, data: dict[str, Any], user_id: int | None = N
# Note: Refill can be disabled, use the existing poller cadence, or run on a user-defined minute interval.
settings['refill_enabled'] = 0 if refill_mode == 'off' else 1
settings['refill_interval_minutes'] = _int_setting(data, current, 'refill_interval_minutes', 5, 1) if refill_mode == 'custom' else 0
# Note: Surge refill is a separate periodic over-cap starter; it never changes the normal target limit.
settings['surge_refill_enabled'] = 1 if data.get('surge_refill_enabled', current.get('surge_refill_enabled', 0)) else 0
settings['surge_refill_interval_minutes'] = _int_setting(data, current, 'surge_refill_interval_minutes', 1440, 1)
settings['surge_refill_batch_size'] = _int_setting(data, current, 'surge_refill_batch_size', 2000, 1)
now = utcnow()
with connect() as conn:
conn.execute(
'''INSERT INTO smart_queue_settings(profile_id,enabled,max_active_downloads,stalled_seconds,min_speed_bytes,min_seeds,min_peers,ignore_seed_peer,ignore_speed,manage_stopped,cooldown_minutes,stop_batch_size,start_grace_seconds,protect_active_below_cap,prefer_partial_progress,auto_stop_idle,refill_enabled,refill_interval_minutes,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
'''INSERT INTO smart_queue_settings(profile_id,enabled,max_active_downloads,stalled_seconds,min_speed_bytes,min_seeds,min_peers,ignore_seed_peer,ignore_speed,manage_stopped,cooldown_minutes,stop_batch_size,start_grace_seconds,protect_active_below_cap,prefer_partial_progress,auto_stop_idle,refill_enabled,refill_interval_minutes,surge_refill_enabled,surge_refill_interval_minutes,surge_refill_batch_size,updated_at)
VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(profile_id) DO UPDATE SET
enabled=excluded.enabled,
max_active_downloads=excluded.max_active_downloads,
@@ -236,8 +244,11 @@ def save_settings(profile_id: int, data: dict[str, Any], user_id: int | None = N
auto_stop_idle=excluded.auto_stop_idle,
refill_enabled=excluded.refill_enabled,
refill_interval_minutes=excluded.refill_interval_minutes,
surge_refill_enabled=excluded.surge_refill_enabled,
surge_refill_interval_minutes=excluded.surge_refill_interval_minutes,
surge_refill_batch_size=excluded.surge_refill_batch_size,
updated_at=excluded.updated_at''',
(profile_id, settings['enabled'], settings['max_active_downloads'], settings['stalled_seconds'], settings['min_speed_bytes'], settings['min_seeds'], settings['min_peers'], settings['ignore_seed_peer'], settings['ignore_speed'], settings['manage_stopped'], settings['cooldown_minutes'], settings['stop_batch_size'], settings['start_grace_seconds'], settings['protect_active_below_cap'], settings['prefer_partial_progress'], settings['auto_stop_idle'], settings['refill_enabled'], settings['refill_interval_minutes'], now),
(profile_id, settings['enabled'], settings['max_active_downloads'], settings['stalled_seconds'], settings['min_speed_bytes'], settings['min_seeds'], settings['min_peers'], settings['ignore_seed_peer'], settings['ignore_speed'], settings['manage_stopped'], settings['cooldown_minutes'], settings['stop_batch_size'], settings['start_grace_seconds'], settings['protect_active_below_cap'], settings['prefer_partial_progress'], settings['auto_stop_idle'], settings['refill_enabled'], settings['refill_interval_minutes'], settings['surge_refill_enabled'], settings['surge_refill_interval_minutes'], settings['surge_refill_batch_size'], now),
)
return get_settings(profile_id, user_id)
@@ -1152,6 +1163,234 @@ def _refill_underfilled_queue(profile: dict, settings: dict[str, Any], profile_i
'start_source_skipped': len(source_skipped),
'checked': len(torrents),
'excluded': len(user_excluded),
'rtorrent_cap': rtorrent_cap,
'settings': settings,
}
def surge_refill_remaining(settings: dict[str, Any]) -> int:
"""Return seconds until the next over-cap Surge refill may run."""
# Note: Surge refill has its own timer because it intentionally starts more torrents than the normal cap.
if not int(settings.get('surge_refill_enabled') or 0):
return 0
minutes = int(settings.get('surge_refill_interval_minutes') or 0)
if minutes <= 0:
return 0
last = _ts(settings.get('last_surge_refill_at'))
if not last:
return 0
return max(0, int((last + minutes * 60) - time.time()))
def _mark_surge_refill_run(profile_id: int, user_id: int) -> None:
# Note: The over-cap refill timer is updated even when no candidates are found, preventing tight retry loops.
with connect() as conn:
conn.execute('UPDATE smart_queue_settings SET last_surge_refill_at=?, updated_at=? WHERE profile_id=?', (utcnow(), utcnow(), profile_id))
def _surge_refill_over_limit(profile: dict, settings: dict[str, Any], profile_id: int, user_id: int) -> dict[str, Any]:
"""Start a large user-defined batch above the Smart Queue cap, then let normal checks drain it."""
# Note: Surge refill never raises max_active_downloads; it only overfills once per configured interval.
torrents = rtorrent.list_torrents(profile)
user_excluded = _excluded_hashes(profile_id, user_id)
max_active = max(1, int(settings.get('max_active_downloads') or 5))
batch_size = max(1, int(settings.get('surge_refill_batch_size') or 2000))
stalled_label_hashes = {str(t.get('hash') or '') for t in torrents if _has_stalled_label(str(t.get('label') or '')) and t.get('hash')}
downloading = [
t for t in torrents
if _is_running_download_slot(t)
and str(t.get('hash') or '') not in user_excluded
]
stopped = [
t for t in torrents
if str(t.get('hash') or '') not in user_excluded
and str(t.get('hash') or '') not in stalled_label_hashes
and _is_waiting_download_candidate(t, True)
and not _is_running_download_slot(t)
]
if int(settings.get('auto_stop_idle') or 0) and not downloading and not stopped:
idle_details = {
'decision': 'Smart Queue auto-stopped during Surge refill: no active or waiting downloads',
'enabled': False,
'auto_stop_idle': True,
'surge_refill': True,
'checked': len(torrents),
'active_before': 0,
'active_after_stop': 0,
'active_after_expected': 0,
'max_active_downloads': max_active,
'surge_refill_batch_size': batch_size,
'over_limit': 0,
'stopped': [],
'started': [],
'start_requested': [],
'stalled_detected': 0,
'stalled_stopped': 0,
'protected_stalled': 0,
'excluded': len(user_excluded),
'excluded_stalled': len(stalled_label_hashes),
}
_mark_surge_refill_run(profile_id, user_id)
_diagnostics_write('smart_queue.surge_refill_idle', {'profile_id': profile_id, 'checked': len(torrents)}, idle_details)
return _disable_when_idle(profile_id, user_id, torrents, idle_details)
startable_stopped, source_skipped = _split_start_candidates(stopped)
prefer_partial_progress = bool(int(settings.get('prefer_partial_progress', 1) or 0))
candidates = sorted(
startable_stopped,
key=lambda t: _start_candidate_sort_key(t, prefer_partial_progress),
reverse=True,
)
c = rtorrent.client_for(profile)
rtorrent_cap = _ensure_rtorrent_download_cap(c, max(max_active, len(downloading) + batch_size))
label_failed: list[str] = []
to_start = candidates[:batch_size]
to_label_waiting = candidates[batch_size:]
for t in to_label_waiting:
h = str(t.get('hash') or '')
if not h:
continue
try:
if not _mark_auto_stopped(c, profile_id, t):
label_failed.append(h)
except Exception:
label_failed.append(h)
start_summary = _start_and_verify_downloads(c, profile_id, to_start)
active_verified = start_summary['active_verified']
start_pending_confirmation = start_summary.get('start_pending_confirmation', [])
start_failed = start_summary['start_failed']
start_requested = start_summary['start_requested']
start_results = start_summary['start_results']
_record_start_grace(profile_id, start_requested)
for h in start_requested:
_restore_auto_label(c, profile_id, h, None)
try:
rtorrent.clear_post_check_download_label(c, h, None)
except Exception:
label_failed.append(h)
keep_labels = (
{str(t.get('hash') or '') for t in to_label_waiting}
| {str(t.get('hash') or '') for t in stopped if _has_smart_queue_label(str(t.get('label') or '')) and str(t.get('hash') or '') not in set(start_requested)}
)
restored = _cleanup_auto_labels(c, profile_id, torrents, keep_labels, True)
active_transferring = sum(1 for t in downloading if int(t.get('down_rate') or 0) > 0 or int(t.get('up_rate') or 0) > 0)
active_rtorrent = sum(1 for t in downloading if int(t.get('active') or 0))
active_state = sum(1 for t in downloading if int(t.get('state') or 0))
active_after_expected = len(downloading) + len(start_requested)
over_limit_expected = max(0, active_after_expected - max_active)
if start_requested:
decision = f'Surge refill requested {len(start_requested)} over-cap start(s); normal checks will drain overflow'
blocked_reason = ''
elif not candidates:
decision = 'Surge refill skipped: no stopped candidates available'
blocked_reason = 'no_candidates'
else:
decision = 'Surge refill ran but rTorrent did not confirm new starts yet'
blocked_reason = 'start_not_confirmed'
details = {
'decision': decision,
'blocked_reason': blocked_reason,
'enabled': bool(settings.get('enabled')),
'surge_refill': True,
'surge_refill_interval_minutes': int(settings.get('surge_refill_interval_minutes') or 0),
'surge_refill_batch_size': batch_size,
'active_before': len(downloading),
'active_after_expected': active_after_expected,
'active_transferring_count': active_transferring,
'active_rtorrent_count': active_rtorrent,
'active_state_count': active_state,
'max_active_downloads': max_active,
'over_limit': over_limit_expected,
'candidates': len(candidates),
'started_planned': len(to_start),
'waiting_labeled': len(to_label_waiting),
'start_requested': start_requested,
'start_results': start_results,
'active_verified_count': len(active_verified),
'pending_confirmation_count': len(start_pending_confirmation),
'start_pending_confirmation': start_pending_confirmation,
'start_failed': start_failed,
'labels_failed': label_failed,
'labels_restored': restored,
'start_source_skipped': len(source_skipped),
'rtorrent_cap_updated': bool(rtorrent_cap.get('updated')),
'rtorrent_cap': rtorrent_cap,
'excluded': len(user_excluded),
'excluded_stalled': len(stalled_label_hashes),
}
_diagnostics_write(
'smart_queue.surge_refill',
{
'profile_id': profile_id,
'checked': len(torrents),
'active_before': len(downloading),
'active_after_expected': active_after_expected,
'max_active_downloads': max_active,
'over_limit': over_limit_expected,
'batch_size': batch_size,
'candidates': len(candidates),
'requested': len(start_requested),
'verified': len(active_verified),
'pending': len(start_pending_confirmation),
'start_failed': len(start_failed),
'waiting_labeled': len(to_label_waiting),
'blocked_reason': blocked_reason,
'rtorrent_cap_updated': bool(rtorrent_cap.get('updated')),
},
{
'rtorrent_cap': rtorrent_cap,
'settings': {
'surge_refill_interval_minutes': int(settings.get('surge_refill_interval_minutes') or 0),
'surge_refill_batch_size': batch_size,
'prefer_partial_progress': prefer_partial_progress,
},
'to_start': _diagnostics_torrents(to_start),
'to_label_waiting': _diagnostics_torrents(to_label_waiting),
'source_skipped': _diagnostics_torrents(source_skipped),
'pending_confirmation': _diagnostics_sample(start_pending_confirmation),
'start_failed': _diagnostics_sample(start_failed),
'labels_failed': _diagnostics_sample(label_failed),
},
)
_mark_surge_refill_run(profile_id, user_id)
add_history(profile_id, 'surge_refill', [], start_requested, len(torrents), details, user_id)
settings = get_settings(profile_id, user_id)
return {
'ok': True,
'enabled': bool(settings.get('enabled')),
'surge_refill': True,
'cooldown_skipped': True,
'refill_mode': _refill_mode(settings),
'refill_remaining_seconds': refill_remaining(settings),
'surge_refill_remaining_seconds': surge_refill_remaining(settings),
'paused': [],
'resumed': start_requested,
'stopped': [],
'started': start_requested,
'start_requested': start_requested,
'start_batch_size': start_summary['start_batch_size'],
'start_verify_attempts': start_summary['start_verify_attempts'],
'start_verify_delay_seconds': start_summary['start_verify_delay_seconds'],
'waiting_labeled': len(to_label_waiting),
'labels_restored': restored,
'labels_failed': label_failed,
'start_failed': start_failed,
'start_no_effect': start_summary['start_no_effect'],
'start_pending_confirmation': start_pending_confirmation,
'active_verified': active_verified,
'active_before': len(downloading),
'active_after_expected': active_after_expected,
'over_limit': over_limit_expected,
'active_transferring_count': active_transferring,
'active_rtorrent_count': active_rtorrent,
'active_state_count': active_state,
'blocked_reason': blocked_reason,
'start_source_skipped': len(source_skipped),
'checked': len(torrents),
'excluded': len(user_excluded),
'settings': settings,
}
@@ -1177,13 +1416,18 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool =
profile_id = int(profile['id'])
settings = get_settings(profile_id, user_id)
remaining = cooldown_remaining(settings)
if not force and int(settings.get('enabled') or 0) and int(settings.get('surge_refill_enabled') or 0) and not surge_refill_remaining(settings):
try:
return _surge_refill_over_limit(profile, settings, profile_id, user_id)
except Exception as exc:
return {'ok': True, 'enabled': True, 'surge_refill': False, 'settings': settings, 'error': str(exc)}
if remaining and not force:
if int(settings.get('enabled') or 0):
refill_wait = refill_remaining(settings)
if not int(settings.get('refill_enabled') or 0):
return {'ok': True, 'enabled': True, 'cooldown_skipped': True, 'cooldown_refill': False, 'refill_disabled': True, 'cooldown_remaining_seconds': remaining, 'settings': settings}
return {'ok': True, 'enabled': True, 'cooldown_skipped': True, 'cooldown_refill': False, 'refill_disabled': True, 'cooldown_remaining_seconds': remaining, 'surge_refill_remaining_seconds': surge_refill_remaining(settings), 'settings': settings}
if refill_wait:
return {'ok': True, 'enabled': True, 'cooldown_skipped': True, 'cooldown_refill': False, 'refill_wait_seconds': refill_wait, 'cooldown_remaining_seconds': remaining, 'settings': settings}
return {'ok': True, 'enabled': True, 'cooldown_skipped': True, 'cooldown_refill': False, 'refill_wait_seconds': refill_wait, 'cooldown_remaining_seconds': remaining, 'surge_refill_remaining_seconds': surge_refill_remaining(settings), 'settings': settings}
try:
# Note: Cooldown still blocks the full Smart Queue pass, but configured refill may fill free slots safely.
refill = _refill_underfilled_queue(profile, settings, profile_id, user_id)
@@ -1191,7 +1435,7 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool =
return refill
except Exception as exc:
return {'ok': True, 'enabled': True, 'cooldown_skipped': True, 'cooldown_refill': False, 'cooldown_remaining_seconds': remaining, 'settings': settings, 'error': str(exc)}
return {'ok': True, 'enabled': bool(settings.get('enabled')), 'cooldown_skipped': True, 'cooldown_remaining_seconds': remaining, 'settings': settings}
return {'ok': True, 'enabled': bool(settings.get('enabled')), 'cooldown_skipped': True, 'cooldown_remaining_seconds': remaining, 'surge_refill_remaining_seconds': surge_refill_remaining(settings), 'settings': settings}
if not force and not int(settings.get('enabled') or 0):
restored: list[str] = []
try:
@@ -1534,4 +1778,4 @@ def check(profile: dict | None = None, user_id: int | None = None, force: bool =
mark_run(profile_id, user_id)
settings = get_settings(profile_id, user_id)
remaining = cooldown_remaining(settings)
return {'ok': True, 'enabled': bool(settings.get('enabled')), 'paused': stopped_by_queue, 'resumed': started_by_queue, 'stopped': stopped_by_queue, 'started': started_by_queue, 'start_requested': start_requested, 'start_batch_size': start_summary['start_batch_size'], 'start_verify_attempts': start_summary['start_verify_attempts'], 'start_verify_delay_seconds': start_summary['start_verify_delay_seconds'], 'waiting_labeled': len(to_label_waiting), 'stalled_labeled': stalled_labeled, 'excluded_stalled': len(stalled_label_hashes), 'manual_labeled_running': len(manual_labeled_running), 'labels_restored': restored, 'labels_failed': label_failed, 'stop_failed': stop_failed, 'start_failed': start_failed, 'start_no_effect': start_no_effect, 'start_pending_confirmation': start_pending_confirmation, 'active_verified': active_verified, 'active_before': len(downloading), 'active_after_stop': active_after_stop, 'over_limit': over_limit, 'stop_eligible': len(stop_eligible), 'start_source_skipped': len(source_skipped), 'ignore_seed_peer': ignore_seed_peer, 'ignore_speed': ignore_speed, 'ignored_seed_peer_count': ignored_seed_peer_count if ignore_seed_peer else 0, 'ignored_speed_count': ignored_speed_count if ignore_speed else 0, 'stalled_seconds': stalled_seconds, 'stalled_timer_key': timer_key, 'stop_batch_size': stop_batch_size, 'start_grace_seconds': start_grace_seconds, 'protect_active_below_cap': protect_active_below_cap, 'prefer_partial_progress': prefer_partial_progress, 'auto_stop_idle': bool(int(settings.get('auto_stop_idle') or 0)), 'stalled_replacement_allowed': stalled_replacement_allowed, 'start_grace_protected': len(start_grace_hashes), 'replacement_capacity': replacement_capacity, 'protected_stalled': protected_stalled, 'healthy_active_protected': 0, 'rtorrent_cap': rtorrent_cap, 'checked': len(torrents), 'excluded': len(user_excluded), 'settings': settings, 'cooldown_remaining_seconds': remaining}
return {'ok': True, 'enabled': bool(settings.get('enabled')), 'paused': stopped_by_queue, 'resumed': started_by_queue, 'stopped': stopped_by_queue, 'started': started_by_queue, 'start_requested': start_requested, 'start_batch_size': start_summary['start_batch_size'], 'start_verify_attempts': start_summary['start_verify_attempts'], 'start_verify_delay_seconds': start_summary['start_verify_delay_seconds'], 'waiting_labeled': len(to_label_waiting), 'stalled_labeled': stalled_labeled, 'excluded_stalled': len(stalled_label_hashes), 'manual_labeled_running': len(manual_labeled_running), 'labels_restored': restored, 'labels_failed': label_failed, 'stop_failed': stop_failed, 'start_failed': start_failed, 'start_no_effect': start_no_effect, 'start_pending_confirmation': start_pending_confirmation, 'active_verified': active_verified, 'active_before': len(downloading), 'active_after_stop': active_after_stop, 'over_limit': over_limit, 'stop_eligible': len(stop_eligible), 'start_source_skipped': len(source_skipped), 'ignore_seed_peer': ignore_seed_peer, 'ignore_speed': ignore_speed, 'ignored_seed_peer_count': ignored_seed_peer_count if ignore_seed_peer else 0, 'ignored_speed_count': ignored_speed_count if ignore_speed else 0, 'stalled_seconds': stalled_seconds, 'stalled_timer_key': timer_key, 'stop_batch_size': stop_batch_size, 'start_grace_seconds': start_grace_seconds, 'protect_active_below_cap': protect_active_below_cap, 'prefer_partial_progress': prefer_partial_progress, 'auto_stop_idle': bool(int(settings.get('auto_stop_idle') or 0)), 'stalled_replacement_allowed': stalled_replacement_allowed, 'start_grace_protected': len(start_grace_hashes), 'replacement_capacity': replacement_capacity, 'protected_stalled': protected_stalled, 'healthy_active_protected': 0, 'rtorrent_cap': rtorrent_cap, 'checked': len(torrents), 'excluded': len(user_excluded), 'settings': settings, 'cooldown_remaining_seconds': remaining, 'surge_refill_remaining_seconds': surge_refill_remaining(settings)}