sangamdas's picture
Add runnable Privacy Finality execution-finality demo
09f1b1f
Raw History Blame Contribute Delete
2.06 kB
from __future__ import annotations
import statistics
import time
from privacy_finality.authority import PermitAuthority, WorkloadKey
from privacy_finality.demo import build_support_act
from privacy_finality.ped import ProtectedEnforcementDomain
from privacy_finality.replay_store import SQLiteFinalityStore
from privacy_finality.sink import FinalitySink
from privacy_finality.vaults import SMEVaultSet
def us(ns: int) -> float:
return ns / 1_000.0
def main(n: int = 500) -> None:
validation = []
issuance = []
sink_total = []
vaults = SMEVaultSet.demo()
ped = ProtectedEnforcementDomain(vaults)
authority = PermitAuthority()
workload = WorkloadKey.generate("sme-ai-support-v1")
sink = FinalitySink(
"support-output-gateway-1",
authority.public_key_b64,
vaults.crypto_policy,
SQLiteFinalityStore(),
)
base = build_support_act()
for i in range(n):
act = type(base)(**{**base.__dict__, "nonce": f"bench-{i}"})
t0 = time.perf_counter_ns()
grant = ped.validate(act)
t1 = time.perf_counter_ns()
permit = authority.issue(grant, workload, ttl_seconds=600)
proof = workload.proof(permit)
t2 = time.perf_counter_ns()
sink.release(act, permit, proof, dict(grant.selected_values))
t3 = time.perf_counter_ns()
validation.append(us(t1 - t0))
issuance.append(us(t2 - t1))
sink_total.append(us(t3 - t2))
def line(name, values):
values_sorted = sorted(values)
p95 = values_sorted[int(0.95 * (len(values_sorted) - 1))]
p99 = values_sorted[int(0.99 * (len(values_sorted) - 1))]
print(f"{name:20s} median={statistics.median(values):9.2f} us p95={p95:9.2f} us p99={p99:9.2f} us")
print(f"iterations={n}")
line("PED validation", validation)
line("permit + PoP", issuance)
line("sink verify+commit", sink_total)
print("NOTE: local Python microbenchmark only; not a protocol performance claim.")
if __name__ == "__main__":
main()