Fleet Group: solo mining condiviso da un gruppo

diff --git a/crates/bp-group-solo-engine/src/distribution.rs b/crates/bp-group-solo-engine/src/distribution.rs
index 451b6c6..db8fb6b 100644
--- a/crates/bp-group-solo-engine/src/distribution.rs
+++ b/crates/bp-group-solo-engine/src/distribution.rs
@@ -171,6 +171,22 @@ async fn compute_distribution(
       .ok_or(DistributionError::GroupNotFound { group_id })?;
   let finder_bonus_ppm = group_row.finder_bonus_ppm.unwrap_or(0).max(0) as u32;

+    // SoloFury patch (sf-group-manager-fee): a fixed manager quota for the
+    // group creator, as ppm of the miner cut, read from the optional column
+    // `pplns_group."managerFeePpm"` (added by the operator, not by a
+    // migration). When set (> 0) it REPLACES the finder bonus: the same
+    // score-weight boost `S·m/(1e6−m)` pays exactly `m` of the miner cut to
+    // the creator's address at every revenue, and the session's own finder
+    // bonus is not applied. A missing column or a NULL/0 value leaves
+    // upstream behaviour untouched. Only the recipient and the ppm change;
+    // share validation, templates and the §4 builder are upstream code.
+    let manager_fee_ppm = sf_manager_fee_ppm(pool, group_id).await;
+    let (bonus_ppm, bonus_address): (u32, AddressId) = if manager_fee_ppm > 0 {
+        (manager_fee_ppm, group_row.creator_address.clone())
+    } else {
+        (finder_bonus_ppm, finder_address.clone())
+    };
+
   // 2. Round state from Redis. Mode-aware: a PROP group reads its per-round
   //    aggregate; a Window group trims to the sliding window first, so the
   //    built distribution is always current (even for an idle group).
@@ -196,8 +212,8 @@ async fn compute_distribution(
       fee_percent: config.fee_percent,
       min_payout_sats: config.min_payout_sats,
       coinbase_weight_budget: config.coinbase_weight_budget,
-        finder_bonus_ppm,
-        finder_address: Some(finder_address),
+        finder_bonus_ppm: bonus_ppm,
+        finder_address: Some(&bonus_address),
       // An empty round is routine (every reset DELs the hash). Only the
       // uncached call bootstraps, and builds are per-finder, so a bootstrap
       // distribution neither outlives the empty round nor reaches another
@@ -238,3 +254,26 @@ mod tests {
       assert_eq!(dist_cfg.coinbase_weight_budget, 60_000);
   }
}
+
+/// SoloFury patch: `pplns_group."managerFeePpm"` if the column exists and the
+/// value is positive, else 0. The query is runtime (not a sqlx macro) so the
+/// offline query cache and the upstream schema stay untouched; a missing
+/// column is logged once per process and treated as 0.
+async fn sf_manager_fee_ppm(pool: &PgPool, group_id: Uuid) -> u32 {
+    static WARNED: std::sync::OnceLock<()> = std::sync::OnceLock::new();
+    let r: Result<Option<i32>, sqlx::Error> =
+        sqlx::query_scalar(r#"SELECT "managerFeePpm" FROM pplns_group WHERE id = $1"#)
+            .bind(group_id)
+            .fetch_one(pool)
+            .await;
+    match r {
+        Ok(Some(ppm)) if ppm > 0 => ppm as u32,
+        Ok(_) => 0,
+        Err(err) => {
+            if WARNED.set(()).is_ok() {
+                tracing::warn!(%err, "sf-group-manager-fee: managerFeePpm unreadable (column missing?); using 0");
+            }
+            0
+        }
+    }
+}
diff --git a/crates/bp-group-solo-engine/tests/sf_manager_fee_integration.rs b/crates/bp-group-solo-engine/tests/sf_manager_fee_integration.rs
new file mode 100644
index 0000000..b4be12c
--- /dev/null
+++ b/crates/bp-group-solo-engine/tests/sf_manager_fee_integration.rs
@@ -0,0 +1,189 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+//! SoloFury patch `sf-group-manager-fee`: the group creator receives a fixed
+//! quota of the miner cut, read from `pplns_group."managerFeePpm"`.
+//! Needs Postgres (`BP_PG_URL`) and Redis (`BP_REDIS_URL`); skips otherwise.
+
+use std::time::Duration;
+
+use bp_common::AddressId;
+use bp_group_solo_engine::config::GroupSoloEngineConfig;
+use bp_group_solo_engine::distribution::{DistributionBuilder, DistributionConfig};
+use bp_group_solo_engine::round::GroupRoundStore;
+use redis::Client;
+use sqlx::{postgres::PgPoolOptions, PgPool};
+use uuid::Uuid;
+
+const REDIS_URL: &str = "redis://127.0.0.1:16379";
+const PG_URL: &str = "postgres://postgres:postgres@localhost:15433/public_pool";
+const FEE_ADDR: &str = "3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy";
+const MANAGER: &str = "bc1qrp33g0q5c5txsp9arysrx4k6zdkfs4nce4xj0gdcccefvpysxf3qccfmv3";
+const CLIENT_A: &str = "bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t4";
+const CLIENT_B: &str = "bc1qar0srrr7xfkvy5l643lydnw9re59gtzzwf5mdq";
+const SUBSIDY: u64 = 312_500_000;
+
+struct H {
+    pool: PgPool,
+    builder: DistributionBuilder,
+    round: GroupRoundStore,
+    group_id: Uuid,
+}
+
+async fn spawn_or_skip(redis_db: u8, manager_ppm: Option<i32>, finder_ppm: Option<i32>) -> Option<H> {
+    let pg_url = std::env::var("BP_PG_URL").unwrap_or_else(|_| PG_URL.to_string());
+    let redis_base = std::env::var("BP_REDIS_URL").unwrap_or_else(|_| REDIS_URL.to_string());
+    let pool = match tokio::time::timeout(
+        Duration::from_secs(2),
+        PgPoolOptions::new().max_connections(2).connect(&pg_url),
+    )
+    .await
+    {
+        Ok(Ok(p)) => p,
+        _ => {
+            eprintln!("PG unavailable — skipping");
+            return None;
+        }
+    };
+    // Own slots in the distribution binary's range (20..23 are unused there).
+    let redis_db =
+        bp_test_support::redis_db_in_range(bp_test_support::redis_db::GS_DISTRIBUTION, redis_db).await;
+    let client = Client::open(format!("{redis_base}/{redis_db}")).ok()?;
+    let mut conn = match tokio::time::timeout(
+        Duration::from_secs(2),
+        bp_test_support::connection_manager(client),
+    )
+    .await
+    {
+        Ok(Ok(c)) => c,
+        _ => {
+            eprintln!("Redis unavailable — skipping");
+            return None;
+        }
+    };
+    redis::cmd("FLUSHDB").query_async::<()>(&mut conn).await.ok()?;
+
+    // The operator adds the column outside the migrations (see the patch).
+    sqlx::query(r#"ALTER TABLE pplns_group ADD COLUMN IF NOT EXISTS "managerFeePpm" integer"#)
+        .execute(&pool)
+        .await
+        .expect("add column");
+    let group_id = Uuid::new_v4();
+    sqlx::query(
+        r#"INSERT INTO pplns_group
+             (id, name, "creatorAddress", "adminTokenHash", active, "createdAt", "updatedAt",
+              "isPublic", "finderBonusPpm", "managerFeePpm")
+           VALUES ($1, $2, $3, $4, true, 0, 0, false, $5, $6)"#,
+    )
+    .bind(group_id)
+    .bind(format!("sf-mgr-{group_id}"))
+    .bind(MANAGER)
+    .bind(format!("hash-{group_id}"))
+    .bind(finder_ppm)
+    .bind(manager_ppm)
+    .execute(&pool)
+    .await
+    .expect("seed group");
+
+    let round = GroupRoundStore::new(conn);
+    let cfg = DistributionConfig::from_engine_config(&GroupSoloEngineConfig {
+        fee_address: Some(AddressId::new(FEE_ADDR).unwrap()),
+        ..GroupSoloEngineConfig::default()
+    });
+    let builder = DistributionBuilder::new(pool.clone(), round.clone(), cfg);
+    Some(H { pool, builder, round, group_id })
+}
+
+async fn cleanup(h: &H) {
+    sqlx::query("DELETE FROM pplns_group WHERE id = $1")
+        .bind(h.group_id)
+        .execute(&h.pool)
+        .await
+        .ok();
+}
+
+fn outputs(h_entries: &[(AddressId, u64)], a: &str) -> Vec<i64> {
+    h_entries
+        .iter()
+        .filter(|(x, _)| x.as_str() == a)
+        .map(|(_, s)| *s as i64)
+        .collect()
+}
+
+#[tokio::test]
+async fn manager_quota_is_an_exact_fraction_of_the_miner_cut() {
+    // 5 % of the block with a 1 % pool fee = 50 505 ppm of the miner cut.
+    let Some(h) = spawn_or_skip(20, Some(50_505), None).await else { return };
+    let a = AddressId::new(CLIENT_A).unwrap();
+    let b = AddressId::new(CLIENT_B).unwrap();
+    let g = h.group_id.to_string();
+    h.round.record_share(None, &g, a.as_str(), 75.0, 1).await.unwrap();
+    h.round.record_share(None, &g, b.as_str(), 25.0, 2).await.unwrap();
+
+    let result = h.builder.build(h.group_id, SUBSIDY, &a).await.expect("ok");
+    let entries = result.distribution.payout_entries_at(SUBSIDY).expect("§4 vector");
+    let pot = bp_share::miner_pot_sats(result.distribution.fee_ppm, SUBSIDY) as i64;
+
+    let mgr = outputs(&entries, MANAGER);
+    assert_eq!(mgr.len(), 1, "manager has exactly one output");
+    let expected_mgr = pot * 50_505 / 1_000_000;
+    assert!((mgr[0] - expected_mgr).abs() <= 2, "manager {} expected {}", mgr[0], expected_mgr);
+
+    let rest = pot - mgr[0];
+    let oa = outputs(&entries, CLIENT_A)[0];
+    let ob = outputs(&entries, CLIENT_B)[0];
+    assert!((oa - rest * 3 / 4).abs() <= 2, "client A {} expected {}", oa, rest * 3 / 4);
+    assert!((ob - rest / 4).abs() <= 2, "client B {} expected {}", ob, rest / 4);
+    // The fee output is still exactly the pool fee: nothing of the quota leaks to the pool.
+    let fee = outputs(&entries, FEE_ADDR)[0];
+    assert!((fee - (SUBSIDY as i64 - pot)).abs() <= 2, "fee {} expected {}", fee, SUBSIDY as i64 - pot);
+    cleanup(&h).await;
+}
+
+#[tokio::test]
+async fn manager_quota_replaces_the_finder_bonus_and_is_finder_independent() {
+    let Some(h) = spawn_or_skip(21, Some(20_202), Some(300_000)).await else { return };
+    let a = AddressId::new(CLIENT_A).unwrap();
+    let b = AddressId::new(CLIENT_B).unwrap();
+    let g = h.group_id.to_string();
+    h.round.record_share(None, &g, a.as_str(), 50.0, 1).await.unwrap();
+    h.round.record_share(None, &g, b.as_str(), 50.0, 2).await.unwrap();
+
+    let ra = h.builder.build(h.group_id, SUBSIDY, &a).await.expect("ok");
+    let rb = h.builder.build(h.group_id, SUBSIDY, &b).await.expect("ok");
+    let ea = ra.distribution.payout_entries_at(SUBSIDY).unwrap();
+    let eb = rb.distribution.payout_entries_at(SUBSIDY).unwrap();
+    // Same coinbase whoever the finder is: equal peers get equal outputs, the
+    // 30 % finder bonus of the row is NOT applied.
+    assert_eq!(outputs(&ea, CLIENT_A), outputs(&ea, CLIENT_B));
+    assert_eq!(outputs(&ea, MANAGER), outputs(&eb, MANAGER));
+    assert_eq!(outputs(&ea, CLIENT_A), outputs(&eb, CLIENT_A));
+    cleanup(&h).await;
+}
+
+#[tokio::test]
+async fn zero_or_null_quota_keeps_upstream_behaviour() {
+    let Some(h) = spawn_or_skip(22, None, Some(3_200)).await else { return };
+    let a = AddressId::new(CLIENT_A).unwrap();
+    let b = AddressId::new(CLIENT_B).unwrap();
+    let g = h.group_id.to_string();
+    h.round.record_share(None, &g, a.as_str(), 50.0, 1).await.unwrap();
+    h.round.record_share(None, &g, b.as_str(), 50.0, 2).await.unwrap();
+    let r = h.builder.build(h.group_id, SUBSIDY, &a).await.expect("ok");
+    let e = r.distribution.payout_entries_at(SUBSIDY).unwrap();
+    assert!(outputs(&e, MANAGER).is_empty(), "no manager output without a quota");
+    assert!(outputs(&e, CLIENT_A)[0] > outputs(&e, CLIENT_B)[0], "upstream finder bonus still applies");
+    cleanup(&h).await;
+}
+
+#[tokio::test]
+async fn empty_round_bootstrap_still_pays_the_manager_quota() {
+    let Some(h) = spawn_or_skip(23, Some(50_505), None).await else { return };
+    let a = AddressId::new(CLIENT_A).unwrap();
+    let r = h.builder.build(h.group_id, SUBSIDY, &a).await.expect("ok");
+    let e = r.distribution.payout_entries_at(SUBSIDY).unwrap();
+    let pot = bp_share::miner_pot_sats(r.distribution.fee_ppm, SUBSIDY) as i64;
+    let mgr = outputs(&e, MANAGER);
+    assert_eq!(mgr.len(), 1);
+    assert!((mgr[0] - pot * 50_505 / 1_000_000).abs() <= 2);
+    assert!((outputs(&e, CLIENT_A)[0] - (pot - mgr[0])).abs() <= 2, "bootstrap claimant gets the rest");
+    cleanup(&h).await;
+}

# blitzpool-sf test run — 2026-10-24T00:29Z — base d10ab0e + sf-group-manager-fee (cb06164)
rustc rustc 1.97.0 (2d8144b78 2026-07-07) · Postgres 16 (migrations 0000-0022) · Redis 8 (1024 db)
   Running unittests src/lib.rs (target/debug/deps/bp_group_solo_engine-2d27263d2fe5c576)
test config::tests::a_non_zero_fee_validates ... ok
test config::tests::a_typo_in_the_fee_address_is_refused ... ok
test config::tests::default_validates_clean ... ok
test config::tests::fee_percent_above_hundred_rejects ... ok
test config::tests::fee_percent_nan_rejects ... ok
test config::tests::fee_percent_negative_rejects ... ok
test config::tests::min_payout_below_dust_rejects ... ok
test config::tests::min_payout_exactly_dust_accepts ... ok
test config::tests::the_default_config_is_refused_because_it_has_no_fee_address ... ok
test config::tests::weight_budget_too_low_rejects ... ok
test distribution::tests::distribution_config_from_engine_config_carries_fields ... ok
test engine::tests::engine_error_carries_source_variants ... ok
test engine::tests::history_rows_transcribe_every_paid_output_besides_the_pool ... ok
test engine::tests::lookup_error_reuses_cached_mode_never_misroutes_window ... ok
test engine::tests::trim_watermark_gates_to_new_buckets_only ... ok
test reset::tests::custom_across_the_dst_fall_back_fires_at_local_midnight ... ok
test reset::tests::daily_on_the_dst_fall_back_day_moves_to_the_next_day ... ok
test reset::tests::next_fire_custom_no_last_reset_fires_at_next_midnight ... ok
test reset::tests::next_fire_custom_with_recent_last_reset_skips_until_interval_elapsed ... ok
test reset::tests::next_fire_daily_in_utc ... ok
test reset::tests::next_fire_daily_in_zurich_tz ... ok
test reset::tests::next_fire_monthly_lands_on_first ... ok
test reset::tests::next_fire_on_a_day_without_midnight_does_not_panic ... ok
test reset::tests::next_fire_weekly_lands_on_monday ... ok
test reset::tests::preset_from_wire_strings ... ok
test reset::tests::reset_runner_is_cloneable_without_c_clone_bound ... ok
test reset::tests::reset_schedule_from_row_handles_missing_fields ... ok
test reset::tests::weekly_across_the_dst_fall_back_lands_on_monday ... ok
test round::tests::best_share_partial_eq ... ok
test round::tests::key_helpers_format_correctly ... ok
test round::tests::round_stats_total_is_sum_of_per_address ... ok
test round::tests::window_key_helpers_format_correctly ... ok
test reset::tests::monthly_across_the_dst_fall_back_lands_on_the_first ... ok
test result: ok. 33 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.01s
   Running tests/distribution_integration.rs (target/debug/deps/distribution_integration-94ee904bdefb95c1)
test an_empty_round_pays_the_finder_not_the_whole_block_to_the_pool ... ok
test build_for_nonexistent_group_returns_group_not_found ... ok
test a_share_after_a_bootstrap_reaches_the_next_build ... ok
test build_with_shares_returns_payouts_and_writes_snapshot ... ok
test concurrent_same_finder_builds_share_one_compute ... ok
test finder_bonus_from_db_row_is_applied ... ok
test distinct_rewards_for_same_group_finder_run_independently ... ok
test invalidate_all_triggers_fresh_compute ... ok
test result: ok. 8 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.13s
   Running tests/engine_integration.rs (target/debug/deps/engine_integration-7a149d8fac6cb6fa)
test a_group_coinbase_below_the_block_subsidy_is_refused ... ok
test a_group_block_far_off_the_reference_is_still_booked ... ok
test a_redelivered_apply_books_nothing ... ok
test a_redelivered_apply_does_not_reset_the_round_again ... ok
test a_retry_completes_a_reset_the_first_apply_did_not_reach ... ok
test a_richer_block_leaves_nobody_owing ... ok
test block_found_books_the_job_coinbase_after_the_round_moved ... ok
test build_distribution_returns_payouts_after_shares ... ok
test concurrent_applies_of_one_block_book_it_once ... ok
test duplicate_block_found_does_not_double_the_history ... ok
test manual_reset_wipes_group_state ... ok
test on_block_found_applies_distribution_and_resets_round ... ok
test on_block_found_keeps_round_when_reset_flag_false ... ok
test reader_best_difficulty_after_shares ... ok
test on_block_found_with_finder_bonus_merges_duplicate_outputs ... ok
test record_reject_updates_round_rejected_total ... ok
test record_share_then_round_stats_sees_it ... ok
test reschedule_group_arms_and_tears_down_reset_cron ... ok
test spawn_core_skips_startup_reset_crons ... ok
test window_grow_invalidates_mode_cache_keeps_in_window_bucket ... ok
test window_mode_kick_drops_the_member_from_the_payout_source ... ok
test window_mode_record_path_trims_aged_buckets ... ok
test window_mode_reject_is_windowed_not_tallied ... ok
test result: ok. 23 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.56s
   Running tests/regtest_group_solo_block_submit.rs (target/debug/deps/regtest_group_solo_block_submit-22aa8a04b030d4a0)
test group_solo_three_member_distribution_block_accepted_by_core ... ok
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s
   Running tests/reset_integration.rs (target/debug/deps/reset_integration-381fc545f8d05a69)
test reset_custom_preset_skips_if_interval_not_elapsed ... ok
test reset_custom_preset_fires_when_interval_elapsed ... ok
test reset_debounce_skips_within_60_seconds ... ok
test reset_fires_past_debounce_window ... ok
test reset_nonexistent_group_returns_error ... ok
test reset_scheduled_wipes_redis_pg_state_and_stamps ... ok
test reset_one_group_does_not_affect_another ... ok
test result: ok. 7 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.12s
   Running tests/round_integration.rs (target/debug/deps/round_integration-0ffcfc32bf229e96)
test a_redelivered_share_is_still_deduped_across_a_round_reset ... ok
test forget_member_subtracts_contribution ... ok
test multiple_groups_redis_state_is_isolated ... ok
test forget_member_window_removes_the_address_from_both_lanes ... ok
test read_by_address_returns_aggregate ... ok
test read_payout_shares_window_trims_on_read ... ok
test record_reject_increments_per_address ... ok
test read_round_stats_returns_per_address_and_rejected ... ok
test record_share_writes_all_keys ... ok
test record_share_is_idempotent_per_share_id ... ok
test reset_full_wipes_everything_including_last_accepted ... ok
test reset_for_block_found_preserves_last_accepted_share_at ... ok
test update_best_share_only_replaces_on_improvement ... ok
test round_stats_window_reads_the_reject_lane_not_the_running_tally ... ok
test windowed_record_is_idempotent_per_share_id ... ok
test windowed_record_aggregates_into_buckets ... ok
test windowed_reject_lands_in_its_bucket_and_window_aggregate ... ok
test windowed_trim_and_reset_cover_the_reject_lane ... ok
test windowed_trim_does_not_strand_a_negative_entry ... ok
test windowed_trim_drops_aged_buckets ... ok
test result: ok. 20 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.03s
   Running tests/sf_manager_fee_integration.rs (target/debug/deps/sf_manager_fee_integration-2f0a9f24a1978b6e)
test manager_quota_is_an_exact_fraction_of_the_miner_cut ... ok
test empty_round_bootstrap_still_pays_the_manager_quota ... ok
test zero_or_null_quota_keeps_upstream_behaviour ... ok
test manager_quota_replaces_the_finder_bonus_and_is_finder_independent ... ok
test result: ok. 4 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.07s
   Running tests/stream_equivalence.rs (target/debug/deps/stream_equivalence-1186db8b4e5fe73c)
test in_process_and_stream_paths_leave_identical_group_solo_round ... ok
test duplicate_entries_in_the_stream_do_not_break_group_solo_equivalence ... ok
test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.10s
test result: ok. 0 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s