The programming model.

use grafos_collections::vec::FabricVec;
use grafos_collections::map::FabricHashMap;
use grafos_collections::queue::FabricQueue;

// Growable array in fabric memory
let mut v: FabricVec = FabricVec::new(mem_lease, 1024)?;
v.push(&42)?;

// Hash map (open addressing, linear probing)
let mut m: FabricHashMap = FabricHashMap::new(mem_lease, 256)?;
m.put(&1, &100)?;

// Ring buffer for producer-consumer (bounded SPSC)
let mut q: FabricQueue = FabricQueue::new(mem_lease, 64, 16)?;
q.push(&42)?;

let mut kv = KvBuilder::new()
    .hot_buckets(64)
    .default_ttl_secs(300)
    .build()?;

kv.put(b"session:abc", b"user_data")?;
let data = kv.get(b"session:abc")?;
// Key expires when TTL fires — no eviction logic needed

let mut mgr = TopicManager::new();
mgr.create("events", TopicConfig {
    partitions: 4, ..Default::default()
})?;

let mut producer = Producer::new("events");
producer.send(&mut mgr, b"order.created")?;

let mut consumer = Consumer::new("events", "order-processor");
consumer.assign(&[0, 1, 2, 3]);
let messages = consumer.poll(&mgr, 100)?;

// With proc macro:
#[grafos_rpc_service]
pub trait Greeter {
    fn greet(&self, name: String) -> String;
    fn add(&self, a: f64, b: f64) -> f64;
}

// Client call — shared fabric memory, not TCP
let client = GreeterClient::new(transport);
let greeting = client.greet("fabric".to_string())?;

let mut graph = TaskGraph::new();

let extract = graph.add_task(TaskDef::new("extract", extract_fn)
    .output("raw_data"));
let transform = graph.add_task(TaskDef::new("transform", transform_fn)
    .input("raw_data").output("clean_data"));
let load = graph.add_task(TaskDef::new("load", load_fn)
    .input("clean_data"));

graph.add_dependency(extract, transform)?;
graph.add_dependency(transform, load)?;

let plan = graph.build()?; // Topological sort → 3 waves
let result = Executor::run(plan)?;

let result = Pipeline::from_source(VecSource::new(vec![1, 2, 3, 4, 5]))
    .map(|x| x * 2)
    .filter(|x| *x > 4)
    .collect()
    .run()?;
// [6, 8, 10]

let a = FabricTensor::from_data(
    vec![1.0, 2.0, 3.0, 4.0], vec![2, 2])?;
let b = FabricTensor::from_data(
    vec![5.0, 6.0, 7.0, 8.0], vec![2, 2])?;
let c = a.matmul(&b)?;
let d = c.relu()?;

let mut fs = FabricFs::format(vec![block_lease])?;

let mut fh = fs.create("/model/weights.bin")?;
fs.write(&mut fh, &weight_data)?;
fs.close(fh)?;

let fh = fs.open("/model/weights.bin", OpenFlags::Read)?;
let mut buf = vec![0u8; weight_data.len()];
fs.read(&fh, &mut buf)?;

let mut store = TieredObjectStore::new(
    hot_store, cold_store, lru_capacity)?;
let uri: FabricUri = "fabric://default/models/v3".parse()?;
store.put(&uri, &model_bytes, None)?;
// Hot objects in memory, cold on block — automatic tiering

// Mutex — RAII guard, auto-release on drop or crash
let mtx = FabricMutex::new(mem_lease, 0, 0u64)?;
{
    let mut guard = mtx.lock(owner_id, timeout_ms)?;
    *guard += 1;
} // guard drops → lock released

// Barrier — multi-party phase sync
let barrier = FabricBarrier::new(mem_lease, 0, 3)?;
barrier.arrive_and_wait(party_id, timeout_ms)?;

let (mut writer, reader) = RegistryBuilder::new()
    .capacity(64).build()?;

writer.register(
    ServiceRegistration::new("inference-server", "2.1.0", node_id)
        .with_endpoint(ServiceEndpoint::net([10, 0, 0, 5], 8080))
        .with_tag("gpu", "h100")
)?;

let services = reader.lookup("inference-server")?;

// Cross-failure-domain replicated resources
let registry = ReplicatedServiceRegistry::new(
    LogicalResourceName::new("inference"),
    placement_policy(FailureDomain::Region { ... }),
)?;

// Quorum, freshness, and recovery semantics are explicit
let log: ReplicatedLog = ReplicatedLog::with_quorum(
    QuorumPolicy::Majority,
    FreshnessPolicy::ReadIndex,
)?;

// Failure drills exercise the contract surface
let drill = FailureDrill::AvailabilityZoneLoss;
graph.simulate(drill)?;