@@ -57,7 +57,7 @@ use driven_vss::{VssMode, VssProvider};
5757use crate :: executor:: { Executor , OpOutcome } ;
5858use crate :: hooks:: { CommandRunner , HookKind , NoopCommandRunner } ;
5959use crate :: network:: { NetworkProbe , NetworkState , ServiceHealth , ServiceName } ;
60- use crate :: pacer:: PacerCeilings ;
60+ use crate :: pacer:: { Pacer , PacerCeilings } ;
6161use crate :: state:: { ActivityLevel , NewActivity , SourceRow , StateRepo } ;
6262use crate :: time:: Clock ;
6363use crate :: types:: {
@@ -189,6 +189,40 @@ pub struct OrchestratorConfig {
189189 pub post_backup_hook : Option < String > ,
190190 /// How long a hook command may run before it is killed, in seconds.
191191 pub hook_timeout_secs : u32 ,
192+ /// What [`skip_on_metered`](Self::skip_on_metered) DOES on a metered
193+ /// network (V2 metered pause-or-throttle, DESIGN s17): fully pause
194+ /// (default, V1 behaviour) or keep syncing at a reduced bandwidth cap.
195+ pub metered_mode : MeteredMode ,
196+ /// The bandwidth cap (Mbps) applied while metered in
197+ /// [`MeteredMode::Throttle`]. `None` falls back to the normal
198+ /// [`bandwidth_cap_mbps`](Self::bandwidth_cap_mbps).
199+ pub metered_bandwidth_cap_mbps : Option < u32 > ,
200+ }
201+
202+ /// What Driven does on a metered network when
203+ /// [`skip_on_metered`](OrchestratorConfig::skip_on_metered) is on (V2, DESIGN
204+ /// s17).
205+ #[ derive( Debug , Clone , Copy , PartialEq , Eq , Serialize , Deserialize , Default ) ]
206+ #[ serde( rename_all = "snake_case" ) ]
207+ pub enum MeteredMode {
208+ /// Pause sync entirely while metered (the V1 behaviour).
209+ #[ default]
210+ Pause ,
211+ /// Keep syncing but cap bandwidth at
212+ /// [`metered_bandwidth_cap_mbps`](OrchestratorConfig::metered_bandwidth_cap_mbps).
213+ Throttle ,
214+ }
215+
216+ /// The effective bandwidth cap (Mbps) for the current network (V2 metered
217+ /// throttle, DESIGN s17). On a metered network with `skip_on_metered` on and
218+ /// [`MeteredMode::Throttle`], the metered cap applies (falling back to the base
219+ /// cap if unset); otherwise the base cap. Pure, for testability.
220+ fn effective_bandwidth_cap_mbps ( cfg : & OrchestratorConfig , on_metered : bool ) -> Option < u32 > {
221+ if on_metered && cfg. skip_on_metered && cfg. metered_mode == MeteredMode :: Throttle {
222+ cfg. metered_bandwidth_cap_mbps . or ( cfg. bandwidth_cap_mbps )
223+ } else {
224+ cfg. bandwidth_cap_mbps
225+ }
192226}
193227
194228impl Default for OrchestratorConfig {
@@ -208,6 +242,8 @@ impl Default for OrchestratorConfig {
208242 pre_backup_hook : None ,
209243 post_backup_hook : None ,
210244 hook_timeout_secs : 60 ,
245+ metered_mode : MeteredMode :: Pause ,
246+ metered_bandwidth_cap_mbps : None ,
211247 }
212248 }
213249}
@@ -373,6 +409,11 @@ pub struct SyncOrchestrator {
373409 /// [`Self::with_command_runner`]. The hook COMMANDS come from
374410 /// [`OrchestratorConfig`]; this is only the seam that runs them.
375411 command_runner : Arc < dyn CommandRunner > ,
412+ /// The executor's rate pacer, shared in for the V2 metered throttle
413+ /// (DESIGN s17): the orchestrator lowers / lifts its bandwidth cap as the
414+ /// network goes on / off metered. `None` (the default / tests) disables the
415+ /// runtime throttle; the cap then stays at its construction value.
416+ pacer : Option < Arc < dyn Pacer > > ,
376417 /// Per-orchestrator record-at-create ledger (P1-A). The recorder hook wired
377418 /// into the provider by [`Self::with_vss`] pushes each freshly-created
378419 /// shadow GUID here synchronously; `record_vss_orphans` drains it into the
@@ -438,6 +479,7 @@ impl SyncOrchestrator {
438479 shutdown_rx,
439480 vss : None ,
440481 command_runner : Arc :: new ( NoopCommandRunner ) ,
482+ pacer : None ,
441483 vss_create_ledger : Arc :: new ( std:: sync:: Mutex :: new ( std:: collections:: HashMap :: new ( ) ) ) ,
442484 orphan_cleanup_done : Mutex :: new ( false ) ,
443485 suspended : std:: sync:: atomic:: AtomicBool :: new ( false ) ,
@@ -474,6 +516,28 @@ impl SyncOrchestrator {
474516 self
475517 }
476518
519+ /// Share in the executor's [`Pacer`] so the orchestrator can drive the V2
520+ /// metered throttle (DESIGN s17). Pass the SAME `Arc` the executor holds so
521+ /// a runtime cap change is seen by the upload path. Without this the metered
522+ /// throttle is inert (the pacer keeps its construction-time cap).
523+ pub fn with_pacer ( mut self , pacer : Arc < dyn Pacer > ) -> Self {
524+ self . pacer = Some ( pacer) ;
525+ self
526+ }
527+
528+ /// Apply the effective bandwidth cap for the current network to the shared
529+ /// pacer (V2 metered throttle, DESIGN s17). On a metered network in
530+ /// [`MeteredMode::Throttle`] the metered cap applies; otherwise the normal
531+ /// cap. A no-op when no pacer was shared in. Idempotent (the pacer ignores
532+ /// an unchanged rate), so it is safe to call every cycle.
533+ async fn apply_bandwidth_cap ( & self ) {
534+ let Some ( pacer) = & self . pacer else { return } ;
535+ let cfg = self . config . read ( ) . await ;
536+ let on_metered = self . power . current ( ) . await . on_metered_network ;
537+ let mbps = effective_bandwidth_cap_mbps ( & cfg, on_metered) ;
538+ pacer. set_bandwidth_cap ( mbps. map ( f64:: from) ) ;
539+ }
540+
477541 /// Run a configured pre/post backup hook command and record the outcome as
478542 /// an activity row. Returns whether the command SUCCEEDED (clean zero
479543 /// exit). Passes `DRIVEN_HOOK` (`pre`/`post`), `DRIVEN_ACCOUNT_ID`, and -
@@ -797,8 +861,11 @@ impl SyncOrchestrator {
797861 return GateDecision :: Pause ( pause_reason_for_network ( net) ) ;
798862 }
799863
800- // Metered network (DESIGN s5.7): pause if configured.
801- if cfg. skip_on_metered && power. on_metered_network {
864+ // Metered network (DESIGN s5.7): in Pause mode pause; in Throttle mode
865+ // (V2, DESIGN s17) keep syncing - the reduced cap is applied to the
866+ // pacer in `apply_bandwidth_cap` before the source loop.
867+ if cfg. skip_on_metered && power. on_metered_network && cfg. metered_mode == MeteredMode :: Pause
868+ {
802869 return GateDecision :: Pause ( PauseReason :: Metered ) ;
803870 }
804871
@@ -1463,6 +1530,12 @@ impl SyncOrchestrator {
14631530 GateDecision :: Proceed => { }
14641531 }
14651532
1533+ // V2 metered throttle (DESIGN s17): the gates are open, so apply the
1534+ // effective bandwidth cap for the current network to the shared pacer
1535+ // before any upload. On a metered network in Throttle mode this lowers
1536+ // the cap; off metered it lifts it back to the base cap. Idempotent.
1537+ self . apply_bandwidth_cap ( ) . await ;
1538+
14661539 // Remote reconcile phase (DESIGN s5.6): now that the gates are open we
14671540 // may safely issue Drive calls. Guarded to run at most once before the
14681541 // first executing cycle.
@@ -2644,6 +2717,137 @@ mod tests {
26442717 . any( |( k, v) | k == "DRIVEN_RESULT" && v == "ok" ) ) ;
26452718 }
26462719
2720+ /// Records the last `set_bandwidth_cap` argument; the rate gates are no-ops.
2721+ #[ derive( Default ) ]
2722+ struct FakePacer {
2723+ last_cap : StdMutex < Option < Option < f64 > > > ,
2724+ }
2725+
2726+ #[ async_trait]
2727+ impl Pacer for FakePacer {
2728+ async fn permit_request ( & self ) { }
2729+ async fn permit_file_create ( & self ) { }
2730+ async fn permit_bytes ( & self , _n : u64 ) { }
2731+ fn note_response ( & self , _c : crate :: pacer:: ResponseClass ) { }
2732+ fn ceilings ( & self ) -> PacerCeilings {
2733+ PacerCeilings :: default ( )
2734+ }
2735+ fn set_bandwidth_cap ( & self , mbps : Option < f64 > ) {
2736+ * self . last_cap . lock ( ) . unwrap ( ) = Some ( mbps) ;
2737+ }
2738+ }
2739+
2740+ fn power_on_metered ( ) -> PowerState {
2741+ PowerState {
2742+ ac_connected : true ,
2743+ battery_percent : Some ( 100 ) ,
2744+ on_metered_network : true ,
2745+ network_reachable : true ,
2746+ }
2747+ }
2748+
2749+ #[ test]
2750+ fn effective_cap_throttles_only_when_metered_and_throttle_mode ( ) {
2751+ let base = OrchestratorConfig {
2752+ bandwidth_cap_mbps : Some ( 100 ) ,
2753+ skip_on_metered : true ,
2754+ metered_bandwidth_cap_mbps : Some ( 2 ) ,
2755+ ..OrchestratorConfig :: default ( )
2756+ } ;
2757+ let throttle = OrchestratorConfig {
2758+ metered_mode : MeteredMode :: Throttle ,
2759+ ..base. clone ( )
2760+ } ;
2761+ // Off metered -> base cap, regardless of mode.
2762+ assert_eq ! ( effective_bandwidth_cap_mbps( & throttle, false ) , Some ( 100 ) ) ;
2763+ // Metered + throttle -> the metered cap.
2764+ assert_eq ! ( effective_bandwidth_cap_mbps( & throttle, true ) , Some ( 2 ) ) ;
2765+ // Metered + pause -> base cap (it will be paused anyway, not throttled).
2766+ assert_eq ! ( effective_bandwidth_cap_mbps( & base, true ) , Some ( 100 ) ) ;
2767+ // Metered + throttle but no metered cap -> falls back to base.
2768+ let no_cap = OrchestratorConfig {
2769+ metered_bandwidth_cap_mbps : None ,
2770+ ..throttle. clone ( )
2771+ } ;
2772+ assert_eq ! ( effective_bandwidth_cap_mbps( & no_cap, true ) , Some ( 100 ) ) ;
2773+ }
2774+
2775+ #[ tokio:: test]
2776+ async fn metered_throttle_does_not_pause_and_caps_the_pacer ( ) {
2777+ let account = AccountId :: new_v4 ( ) ;
2778+ let dir = tempfile:: tempdir ( ) . unwrap ( ) ;
2779+ let src = source_in ( account, dir. path ( ) ) ;
2780+ let exec = Arc :: new ( RecordingExecutor :: default ( ) ) ;
2781+ let cfg = OrchestratorConfig {
2782+ skip_on_metered : true ,
2783+ metered_mode : MeteredMode :: Throttle ,
2784+ metered_bandwidth_cap_mbps : Some ( 2 ) ,
2785+ ..OrchestratorConfig :: default ( )
2786+ } ;
2787+ let ( orch, _clock) = build (
2788+ account,
2789+ vec ! [ src] ,
2790+ exec. clone ( ) ,
2791+ power_on_metered ( ) ,
2792+ Arc :: new ( FakeNet :: online ( ) ) ,
2793+ cfg,
2794+ ) ;
2795+ let pacer = Arc :: new ( FakePacer :: default ( ) ) ;
2796+ let orch = orch. with_pacer ( pacer. clone ( ) ) ;
2797+
2798+ orch. run_cycle ( TickSource :: Scheduled ) . await . unwrap ( ) ;
2799+
2800+ assert_ne ! (
2801+ orch. state( ) . await ,
2802+ OrchestratorState :: Paused {
2803+ reason: PauseReason :: Metered
2804+ } ,
2805+ "throttle mode must not pause on a metered network"
2806+ ) ;
2807+ assert_eq ! (
2808+ * pacer. last_cap. lock( ) . unwrap( ) ,
2809+ Some ( Some ( 2.0 ) ) ,
2810+ "the metered cap (2 Mbps) was applied to the pacer"
2811+ ) ;
2812+ }
2813+
2814+ #[ tokio:: test]
2815+ async fn metered_pause_mode_still_pauses_and_skips_the_cap ( ) {
2816+ let account = AccountId :: new_v4 ( ) ;
2817+ let dir = tempfile:: tempdir ( ) . unwrap ( ) ;
2818+ let src = source_in ( account, dir. path ( ) ) ;
2819+ let exec = Arc :: new ( RecordingExecutor :: default ( ) ) ;
2820+ let cfg = OrchestratorConfig {
2821+ skip_on_metered : true ,
2822+ metered_mode : MeteredMode :: Pause ,
2823+ ..OrchestratorConfig :: default ( )
2824+ } ;
2825+ let ( orch, _clock) = build (
2826+ account,
2827+ vec ! [ src] ,
2828+ exec. clone ( ) ,
2829+ power_on_metered ( ) ,
2830+ Arc :: new ( FakeNet :: online ( ) ) ,
2831+ cfg,
2832+ ) ;
2833+ let pacer = Arc :: new ( FakePacer :: default ( ) ) ;
2834+ let orch = orch. with_pacer ( pacer. clone ( ) ) ;
2835+
2836+ orch. run_cycle ( TickSource :: Scheduled ) . await . unwrap ( ) ;
2837+
2838+ assert_eq ! (
2839+ orch. state( ) . await ,
2840+ OrchestratorState :: Paused {
2841+ reason: PauseReason :: Metered
2842+ }
2843+ ) ;
2844+ assert_eq ! (
2845+ * pacer. last_cap. lock( ) . unwrap( ) ,
2846+ None ,
2847+ "pause mode pauses before applying any cap"
2848+ ) ;
2849+ }
2850+
26472851 #[ tokio:: test]
26482852 async fn battery_gate_pauses_when_skip_on_battery ( ) {
26492853 // On battery with skip_on_battery => Paused{Battery}, no execute.
0 commit comments