creo (0.1.0)

Published 2026-01-05 21:54:30 +00:00 by kstiehl in kstiehl/creo

Setup this registry in the Cargo configuration file (for example ~/.cargo/config.toml):

[registries.forgejo]
index = "sparse+" # Sparse index
# index = "" # Git

[net]
git-fetch-with-cli = true

To install the package using Cargo, run the following command:

cargo add creo@0.1.0 --registry forgejo

For more information on the Cargo registry, see the documentation.

Creo

Custom Resource Engine + "o" — "I create" in Latin

A Kubernetes CRD-inspired resource storage system for Rust, built on NATS JetStream. Creo provides a distributed, extensible API for persisting objects with built-in optimistic locking, namespaces, multitenancy, and real-time change detection.

Why Creo?

Kubernetes CRDs offer a powerful pattern: an extensible API server where systems can declare, persist, and react to object state changes. Creo brings this pattern to any application using NATS as the distributed storage layer. All this without requiring setting up etcd clusters and being an expert on Kubernetes control planes.

Features

Extensible Object Model: Define custom object types with Kind (group/version/name)
Optimistic Locking: Automatic conflict detection prevents lost updates
Namespaces & Multitenancy: Scope objects by namespace for isolation
Real-Time Updates: Stream changes as they happen via NATS JetStream
Type-Safe API: Generic Object trait with compile-time safety
Distributed by Design: NATS provides clustering, persistence, and pub/sub out of the box

Quick Start

Define an Object Type

use creo::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
struct Task {
    #[serde(skip)]
    meta: Option<Metadata>,
    description: String,
    priority: u32,
}

const TASK_KIND: Kind = Kind {
    group: "workflows",
    name: "task",
    version: "v1",
};

impl Object for Task {
    fn kind() -> &'static Kind { &TASK_KIND }
    fn set_metadata(&mut self, meta: Metadata) { self.meta = Some(meta); }
    fn get_metadata(&self) -> Option<&Metadata> { self.meta.as_ref() }
}

Save and Retrieve Objects

let nats = async_nats::connect("localhost").await?;
let client = Client::new(nats, "my-stream".to_string());

// Create and save
let mut task = Task {
    meta: Some(NamespacedName {
        namespace: "production".into(),
        name: "daily-backup".into(),
    }.into()),
    description: "Run nightly backups".into(),
    priority: 1,
};

client.replace(&mut task).await?;

// List all tasks across all namespaces
let tasks: Vec<Task> = client.list_all().await?;

// List tasks in a specific namespace
let prod_tasks: Vec<Task> = client.list_namespace("production").await?;

Optimistic Locking in Action

// Update with conflict detection
let mut task = tasks[0].clone();
task.priority = 2;
client.apply(&mut task).await?;  // ✓ succeeds

// Concurrent update with stale version
let stale = tasks[0].clone();
client.apply(&mut stale).await?;  // ✗ returns Error::Conflict

Real-Time Change Watching

use futures_util::StreamExt;

// Watch for all changes in a namespace
let mut stream = client
    .watch_namespace::<Task>("production")
    .await?;

// React to changes as they happen
while let Some(result) = stream.next().await {
    match result {
        Ok(task) => {
            println!("Task updated: {} - {}", task.meta.unwrap().namespaced_name, task.description);
            // Process the change...
        }
        Err(e) => eprintln!("Watch error: {}", e),
    }
}

API Overview

Client

impl Client {
    // Save with conflict detection
    async fn apply<O: Object>(&self, object: &mut O) -> Result<()>;

    // Save without conflict detection
    async fn replace<O: Object>(&self, object: &mut O) -> Result<()>;

    // Get by namespaced name
    async fn get<O: Object>(&self, name: &NamespacedName) -> Result<Option<O>>;

    // List all objects across all namespaces
    async fn list_all<O: Object>(&self) -> Result<Vec<O>>;

    // List objects in a specific namespace
    async fn list_namespace<O: Object>(&self, namespace: &str) -> Result<Vec<O>>;

    // List objects (optionally filtered by namespace)
    async fn list<O: Object>(&self, namespace: Option<&str>) -> Result<Vec<O>>;

    // Watch for real-time changes in a namespace
    async fn watch_namespace<O: Object>(&self, namespace: &str) -> Result<impl Stream<Item = Result<O>>>;
}

Core Types

pub struct Kind {
    pub group: &'static str,
    pub name: &'static str,
    pub version: &'static str,
}

pub struct Metadata {
    pub namespaced_name: NamespacedName,
}

pub struct NamespacedName {
    pub namespace: String,
    pub name: String,
}

Requirements

NATS server with JetStream enabled
Rust 2024 edition
Tokio async runtime

License

This project is open source.

Acknowledgments

Inspired by the elegance of Kubernetes Custom Resource Definitions and the power of NATS.

ID	Version
async-nats	^0.45.0
futures-util	^0.3.31
serde	^1.0.228
serde_json	^1.0.148
thiserror	^2.0.17
tokio	^1.48.0

Details

Cargo

kstiehl/creo

2026-01-05 21:54:30 +00:00

18 KiB

Assets (1)

creo-0.1.0.crate 18 KiB

Versions (4) View all

0.3.1

2026-04-03

0.2.1

2026-02-08

0.2.0

2026-01-13

0.1.0

2026-01-05

Issues