use crate::error::{Result, TokenError};
use dashmap::DashMap;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::sync::Arc;
use tenzro_storage::{KvStore, WriteOp, CF_PROVIDERS};
use tenzro_types::primitives::{Address, Timestamp};
use tracing::{debug, info};
const COMPUTE_BOND_VAULT_DOMAIN: &[u8] = b"tenzro/compute-bond/vault";
pub const COMPUTE_BOND_KEY_PREFIX: &[u8] = b"compute_bond:";
pub const DEFAULT_COMPUTE_BOND_COOLDOWN_MS: i64 = 7 * 24 * 60 * 60 * 1000;
pub const DEFAULT_COMPUTE_BOND_MIN: u128 = 100 * 1_000_000_000_000_000_000;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "PascalCase")]
pub enum ComputeBondStatus {
Active,
Cooldown,
Returned,
Slashed,
}
impl ComputeBondStatus {
pub fn as_str(self) -> &'static str {
match self {
ComputeBondStatus::Active => "Active",
ComputeBondStatus::Cooldown => "Cooldown",
ComputeBondStatus::Returned => "Returned",
ComputeBondStatus::Slashed => "Slashed",
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ComputeBondEvent {
Posted { amount: u128, at: Timestamp },
Increased { amount: u128, at: Timestamp },
WithdrawInitiated { cooldown_until: Timestamp, at: Timestamp },
Returned { at: Timestamp },
Slashed {
amount: u128,
remaining: u128,
reason: String,
at: Timestamp,
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ComputeBondState {
pub provider_did: String,
pub provider_address: Address,
pub amount: u128,
pub status: ComputeBondStatus,
pub cooldown_until: Option<Timestamp>,
pub last_modified_block: u64,
#[serde(default)]
pub failure_count: u32,
pub history: Vec<ComputeBondEvent>,
}
impl ComputeBondState {
pub fn is_eligible(&self) -> bool {
matches!(self.status, ComputeBondStatus::Active)
}
pub fn effective_amount(&self) -> u128 {
if self.is_eligible() {
self.amount
} else {
0
}
}
}
pub fn derive_compute_bond_vault_address(provider_did: &str) -> Address {
let mut hasher = Sha256::new();
hasher.update(COMPUTE_BOND_VAULT_DOMAIN);
hasher.update(provider_did.as_bytes());
let digest = hasher.finalize();
let mut bytes = [0u8; 32];
bytes.copy_from_slice(&digest);
Address::new(bytes)
}
pub struct ComputeBondManager {
bonds: DashMap<String, ComputeBondState>,
storage: Option<Arc<dyn KvStore>>,
cooldown_ms: i64,
min_bond: u128,
}
impl Default for ComputeBondManager {
fn default() -> Self {
Self::new()
}
}
impl ComputeBondManager {
pub fn new() -> Self {
Self {
bonds: DashMap::new(),
storage: None,
cooldown_ms: DEFAULT_COMPUTE_BOND_COOLDOWN_MS,
min_bond: DEFAULT_COMPUTE_BOND_MIN,
}
}
pub fn with_storage(storage: Arc<dyn KvStore>) -> Result<Self> {
let mgr = Self {
bonds: DashMap::new(),
storage: Some(storage.clone()),
cooldown_ms: DEFAULT_COMPUTE_BOND_COOLDOWN_MS,
min_bond: DEFAULT_COMPUTE_BOND_MIN,
};
mgr.hydrate_from_storage()?;
Ok(mgr)
}
pub fn with_governance(mut self, cooldown_ms: i64, min_bond: u128) -> Self {
self.cooldown_ms = cooldown_ms;
self.min_bond = min_bond;
self
}
pub fn min_bond(&self) -> u128 {
self.min_bond
}
pub fn cooldown_ms(&self) -> i64 {
self.cooldown_ms
}
fn hydrate_from_storage(&self) -> Result<()> {
let storage = match &self.storage {
Some(s) => s.clone(),
None => return Ok(()),
};
let mut count = 0usize;
let rows = storage
.scan_prefix(CF_PROVIDERS, COMPUTE_BOND_KEY_PREFIX)
.map_err(|e| TokenError::StorageError(format!("scan compute_bond: {}", e)))?;
for (_key, value) in rows {
let bond: ComputeBondState = serde_json::from_slice(&value)
.map_err(|e| TokenError::StorageError(format!("decode compute_bond: {}", e)))?;
self.bonds.insert(bond.provider_did.clone(), bond);
count += 1;
}
info!(count, "ComputeBondManager hydrated from storage");
Ok(())
}
fn persist(&self, bond: &ComputeBondState) -> Result<()> {
if let Some(storage) = &self.storage {
let mut key = COMPUTE_BOND_KEY_PREFIX.to_vec();
key.extend_from_slice(bond.provider_did.as_bytes());
let value = serde_json::to_vec(bond)
.map_err(|e| TokenError::StorageError(format!("encode compute_bond: {}", e)))?;
storage
.write_batch_sync(vec![WriteOp::Put {
cf: CF_PROVIDERS.to_string(),
key,
value,
}])
.map_err(|e| TokenError::StorageError(format!("persist compute_bond: {}", e)))?;
}
Ok(())
}
pub fn post(
&self,
provider_did: &str,
provider_address: Address,
amount: u128,
block_height: u64,
) -> Result<ComputeBondState> {
if amount == 0 {
return Err(TokenError::InvalidParameter(
"compute bond amount must be greater than zero".to_string(),
));
}
if let Some(existing) = self.bonds.get(provider_did) {
if !matches!(existing.status, ComputeBondStatus::Returned) {
return Err(TokenError::InvalidParameter(format!(
"provider {} already has a compute bond (status={})",
provider_did,
existing.status.as_str()
)));
}
}
let now = Timestamp::now();
let state = ComputeBondState {
provider_did: provider_did.to_string(),
provider_address,
amount,
status: ComputeBondStatus::Active,
cooldown_until: None,
last_modified_block: block_height,
failure_count: 0,
history: vec![ComputeBondEvent::Posted { amount, at: now }],
};
self.persist(&state)?;
self.bonds.insert(provider_did.to_string(), state.clone());
debug!(
provider = %provider_did,
amount,
"ComputeBond posted"
);
Ok(state)
}
pub fn increase(
&self,
provider_did: &str,
amount: u128,
block_height: u64,
) -> Result<ComputeBondState> {
if amount == 0 {
return Err(TokenError::InvalidParameter(
"increase amount must be greater than zero".to_string(),
));
}
let mut bond_ref = self
.bonds
.get_mut(provider_did)
.ok_or_else(|| TokenError::NotFound(format!("no compute bond for {}", provider_did)))?;
if bond_ref.status != ComputeBondStatus::Active {
return Err(TokenError::InvalidParameter(format!(
"cannot increase compute bond in {} state",
bond_ref.status.as_str()
)));
}
bond_ref.amount = bond_ref.amount.checked_add(amount).ok_or_else(|| {
TokenError::InvalidParameter("compute bond amount overflow".to_string())
})?;
bond_ref.last_modified_block = block_height;
bond_ref.history.push(ComputeBondEvent::Increased {
amount,
at: Timestamp::now(),
});
let snapshot = bond_ref.clone();
drop(bond_ref);
self.persist(&snapshot)?;
debug!(
provider = %provider_did,
amount,
total = snapshot.amount,
"ComputeBond increased"
);
Ok(snapshot)
}
pub fn withdraw(
&self,
provider_did: &str,
block_height: u64,
) -> Result<ComputeBondState> {
let mut bond_ref = self
.bonds
.get_mut(provider_did)
.ok_or_else(|| TokenError::NotFound(format!("no compute bond for {}", provider_did)))?;
if bond_ref.status != ComputeBondStatus::Active {
return Err(TokenError::InvalidParameter(format!(
"cannot withdraw compute bond in {} state",
bond_ref.status.as_str()
)));
}
let now = Timestamp::now();
let cooldown_until = Timestamp::new(now.0.saturating_add(self.cooldown_ms));
bond_ref.status = ComputeBondStatus::Cooldown;
bond_ref.cooldown_until = Some(cooldown_until);
bond_ref.last_modified_block = block_height;
bond_ref.history.push(ComputeBondEvent::WithdrawInitiated {
cooldown_until,
at: now,
});
let snapshot = bond_ref.clone();
drop(bond_ref);
self.persist(&snapshot)?;
info!(
provider = %provider_did,
cooldown_until = ?cooldown_until,
"ComputeBond withdrawal initiated"
);
Ok(snapshot)
}
pub fn finalize_withdrawal(
&self,
provider_did: &str,
block_height: u64,
) -> Result<ComputeBondState> {
let mut bond_ref = self
.bonds
.get_mut(provider_did)
.ok_or_else(|| TokenError::NotFound(format!("no compute bond for {}", provider_did)))?;
if bond_ref.status != ComputeBondStatus::Cooldown {
return Err(TokenError::InvalidParameter(format!(
"cannot finalize compute bond withdrawal: status is {}",
bond_ref.status.as_str()
)));
}
let cooldown_until = bond_ref.cooldown_until.ok_or_else(|| {
TokenError::InvalidParameter(
"cooldown compute bond missing cooldown_until".to_string(),
)
})?;
if Timestamp::now() < cooldown_until {
return Err(TokenError::InvalidParameter(format!(
"cooldown not yet elapsed (until {:?})",
cooldown_until
)));
}
bond_ref.status = ComputeBondStatus::Returned;
bond_ref.amount = 0;
bond_ref.last_modified_block = block_height;
bond_ref
.history
.push(ComputeBondEvent::Returned { at: Timestamp::now() });
let snapshot = bond_ref.clone();
drop(bond_ref);
self.persist(&snapshot)?;
info!(provider = %provider_did, "ComputeBond returned to provider");
Ok(snapshot)
}
pub fn record_failure(&self, provider_did: &str) -> Result<u32> {
let mut bond_ref = self
.bonds
.get_mut(provider_did)
.ok_or_else(|| TokenError::NotFound(format!("no compute bond for {}", provider_did)))?;
bond_ref.failure_count = bond_ref.failure_count.saturating_add(1);
let new_count = bond_ref.failure_count;
let snapshot = bond_ref.clone();
drop(bond_ref);
self.persist(&snapshot)?;
Ok(new_count)
}
pub fn reset_failure_count(&self, provider_did: &str) -> Result<()> {
let mut bond_ref = self
.bonds
.get_mut(provider_did)
.ok_or_else(|| TokenError::NotFound(format!("no compute bond for {}", provider_did)))?;
if bond_ref.failure_count == 0 {
return Ok(());
}
bond_ref.failure_count = 0;
let snapshot = bond_ref.clone();
drop(bond_ref);
self.persist(&snapshot)?;
Ok(())
}
pub fn slash(
&self,
provider_did: &str,
slash_amount: u128,
reason: &str,
block_height: u64,
) -> Result<ComputeBondState> {
if slash_amount == 0 {
return Err(TokenError::InvalidParameter(
"slash amount must be greater than zero".to_string(),
));
}
let mut bond_ref = self
.bonds
.get_mut(provider_did)
.ok_or_else(|| TokenError::NotFound(format!("no compute bond for {}", provider_did)))?;
if bond_ref.status != ComputeBondStatus::Active {
return Err(TokenError::InvalidParameter(format!(
"cannot slash compute bond in {} state",
bond_ref.status.as_str()
)));
}
let applied = slash_amount.min(bond_ref.amount);
bond_ref.amount = bond_ref.amount.saturating_sub(applied);
if bond_ref.amount == 0 {
bond_ref.status = ComputeBondStatus::Slashed;
}
bond_ref.last_modified_block = block_height;
let remaining = bond_ref.amount;
bond_ref.history.push(ComputeBondEvent::Slashed {
amount: applied,
remaining,
reason: reason.to_string(),
at: Timestamp::now(),
});
let snapshot = bond_ref.clone();
drop(bond_ref);
self.persist(&snapshot)?;
info!(
provider = %provider_did,
slashed = applied,
remaining = snapshot.amount,
reason = %reason,
terminal = matches!(snapshot.status, ComputeBondStatus::Slashed),
"ComputeBond slashed"
);
Ok(snapshot)
}
pub fn get(&self, provider_did: &str) -> Option<ComputeBondState> {
self.bonds.get(provider_did).map(|r| r.clone())
}
pub fn list(&self) -> Vec<ComputeBondState> {
self.bonds.iter().map(|r| r.value().clone()).collect()
}
pub fn effective_for_registration(&self, provider_did: &str) -> u128 {
self.bonds
.get(provider_did)
.map(|r| r.effective_amount())
.unwrap_or(0)
}
pub fn meets_minimum(&self, provider_did: &str) -> bool {
self.effective_for_registration(provider_did) >= self.min_bond
}
}
#[cfg(test)]
mod tests {
use super::*;
fn addr(byte: u8) -> Address {
Address::new([byte; 32])
}
#[test]
fn post_creates_active_bond() {
let m = ComputeBondManager::new();
let bond = m
.post("did:tenzro:provider:p1", addr(1), 5_000, 100)
.unwrap();
assert_eq!(bond.status, ComputeBondStatus::Active);
assert_eq!(bond.amount, 5_000);
assert_eq!(
m.effective_for_registration("did:tenzro:provider:p1"),
5_000
);
}
#[test]
fn double_post_rejects_when_active() {
let m = ComputeBondManager::new();
m.post("p1", addr(1), 100, 1).unwrap();
assert!(m.post("p1", addr(1), 100, 2).is_err());
}
#[test]
fn post_after_returned_succeeds() {
let m = ComputeBondManager::new().with_governance(0, 0);
m.post("p1", addr(1), 100, 1).unwrap();
m.withdraw("p1", 2).unwrap();
m.finalize_withdrawal("p1", 3).unwrap();
assert!(m.post("p1", addr(1), 200, 4).is_ok());
assert_eq!(m.get("p1").unwrap().amount, 200);
}
#[test]
fn increase_only_active() {
let m = ComputeBondManager::new();
m.post("p1", addr(1), 100, 1).unwrap();
m.increase("p1", 50, 2).unwrap();
assert_eq!(m.get("p1").unwrap().amount, 150);
m.withdraw("p1", 3).unwrap();
assert!(m.increase("p1", 50, 4).is_err());
}
#[test]
fn withdraw_demotes_eligibility() {
let m = ComputeBondManager::new();
m.post("p1", addr(1), 100, 1).unwrap();
assert_eq!(m.effective_for_registration("p1"), 100);
m.withdraw("p1", 2).unwrap();
assert_eq!(m.effective_for_registration("p1"), 0);
}
#[test]
fn finalize_rejects_before_cooldown_elapsed() {
let m = ComputeBondManager::new();
m.post("p1", addr(1), 100, 1).unwrap();
m.withdraw("p1", 2).unwrap();
assert!(m.finalize_withdrawal("p1", 3).is_err());
}
#[test]
fn meets_minimum_gate() {
let m = ComputeBondManager::new().with_governance(
DEFAULT_COMPUTE_BOND_COOLDOWN_MS,
500,
);
m.post("p1", addr(1), 400, 1).unwrap();
assert!(!m.meets_minimum("p1"));
m.increase("p1", 200, 2).unwrap();
assert!(m.meets_minimum("p1"));
}
#[test]
fn slash_reduces_amount_and_appends_history() {
let m = ComputeBondManager::new();
m.post("p1", addr(1), 1000, 1).unwrap();
let snap = m.slash("p1", 250, "sla:probe_timeout", 2).unwrap();
assert_eq!(snap.amount, 750);
assert_eq!(snap.status, ComputeBondStatus::Active);
assert!(matches!(
snap.history.last().unwrap(),
ComputeBondEvent::Slashed { amount: 250, remaining: 750, .. }
));
}
#[test]
fn slash_to_zero_transitions_to_slashed_terminal() {
let m = ComputeBondManager::new();
m.post("p1", addr(1), 1000, 1).unwrap();
let snap = m.slash("p1", 1000, "sla:probe_timeout", 2).unwrap();
assert_eq!(snap.amount, 0);
assert_eq!(snap.status, ComputeBondStatus::Slashed);
assert_eq!(m.effective_for_registration("p1"), 0);
assert!(m.slash("p1", 1, "sla:probe_timeout", 3).is_err());
}
#[test]
fn slash_caps_at_balance() {
let m = ComputeBondManager::new();
m.post("p1", addr(1), 500, 1).unwrap();
let snap = m.slash("p1", 999_999, "sla:probe_timeout", 2).unwrap();
assert_eq!(snap.amount, 0);
assert_eq!(snap.status, ComputeBondStatus::Slashed);
}
#[test]
fn slash_rejects_cooldown_and_returned() {
let m = ComputeBondManager::new().with_governance(0, 0);
m.post("p1", addr(1), 100, 1).unwrap();
m.withdraw("p1", 2).unwrap();
assert!(m.slash("p1", 10, "sla:probe_timeout", 3).is_err());
m.finalize_withdrawal("p1", 4).unwrap();
assert!(m.slash("p1", 10, "sla:probe_timeout", 5).is_err());
}
#[test]
fn post_after_slashed_rejects() {
let m = ComputeBondManager::new();
m.post("p1", addr(1), 100, 1).unwrap();
m.slash("p1", 100, "sla:probe_timeout", 2).unwrap();
assert!(m.post("p1", addr(1), 200, 3).is_err());
}
#[test]
fn failure_counter_increments_and_resets() {
let m = ComputeBondManager::new();
m.post("p1", addr(1), 100, 1).unwrap();
assert_eq!(m.record_failure("p1").unwrap(), 1);
assert_eq!(m.record_failure("p1").unwrap(), 2);
assert_eq!(m.record_failure("p1").unwrap(), 3);
m.reset_failure_count("p1").unwrap();
assert_eq!(m.get("p1").unwrap().failure_count, 0);
}
#[test]
fn deterministic_vault_address() {
let a = derive_compute_bond_vault_address("did:tenzro:provider:p1");
let b = derive_compute_bond_vault_address("did:tenzro:provider:p1");
let c = derive_compute_bond_vault_address("did:tenzro:provider:p2");
assert_eq!(a, b);
assert_ne!(a, c);
}
}