Timesource

Event sourcing with Timescaledb as storage engine for Rust applications.

NOTE: public APIs are very unstable. Documentation will be published when API stabilises further.

Features

• Event sourcing with all the benefits that TimescaleDb offers:
  • Partitioned tables for horizontal scalability
  • 10-100x faster queries than postgres
  • Compressed data for lower storage requirements
• Everything can be tracked to the exact moment when it happened. In future, this crate aims to offer APIs for time travelling, and time-aware repositories. All events have a timestamp in nanoseconds.
• Timesource has a flexible approach to event serialisation. It supports either: JSON, CBOR or Protocol Buffers
• Consumers with strong safeguards:
  • Backpressure is applied if the database appears to have issues
  • Slow consumers will self-heal and catch up with events when the rate of new messages is bigger than the time it takes to consume them
  • Offsets are stored in the database so that consumers can carry on where they left off
• Flexible ordering constraints:
  • Events may be published in orderly or unorderly fashion. Yet, consumers will always get events in order.
  • Events within the same aggregate type are ordered by a monotonically increasing integer. Aggregate roots are ordered separately per timestamp.
• The storage engine can handle a total of 9,223,372,036,854,775,807 events up until 11 April 2262 23:47:16.854.

Event serialisation/deserialisation

Timesource supports:

• JSON (using serde json)
• CBOR (using minicbor)
• Protocol Buffers (using prost)

If no configuration is provided, timesource defaults to JSON. Encodings can be configured via derive attributes. For example:

#[derive(serde::Serialize, serde::Deserialize, TimesourceEvent, PartialEq, Debug)]
#[timesource(encoding = "json")] // this is optional, as JSON is already the default
enum TdbEventJson {
    Created,
    Abandoned(String),
    #[allow(dead_code)]
    AddedItem {
        id: usize,
    },
}

#[derive(minicbor::Encode, minicbor::Decode, TimesourceEvent, PartialEq, Debug)]
#[timesource(encoding = "cbor", version = "1.1")] // it's also possible to set the version along with the encoding
enum TdbEventCbor {
    #[b(0)]
    Created,
    #[b(1)]
    Abandoned(#[b(0)] String),
    #[b(2)]
    AddedItem {
        #[b(0)]
        id: usize,
    },
}

mod proto {
    use prost::Message;

    #[derive(Clone, PartialEq, Message)]
    pub struct Created {}
    #[derive(Clone, PartialEq, Message)]
    pub struct Abandoned {
        #[prost(string, tag = "1")]
        pub reason: ::prost::alloc::string::String,
    }
    #[derive(Clone, PartialEq, Message)]
    pub struct AddedItem {
        #[prost(uint32, tag = "1")]
        pub id: u32,
    }
    #[derive(Clone, PartialEq, Message, TimesourceEvent)]
    #[timesource(encoding = "proto")]
    pub struct TdbEvent {
        #[prost(oneof = "tdb_event_proto::Data", tags = "1, 2, 3")]
        pub data: ::core::option::Option<tdb_event_proto::Data>,
    }

    pub mod tdb_event_proto {
        #[derive(Clone, PartialEq, ::prost::Oneof)]
        pub enum Data {
            #[prost(message, tag = "1")]
            Created(super::Created),
            #[prost(message, tag = "2")]
            Abandoned(super::Abandoned),
            #[prost(message, tag = "3")]
            AddedItem(super::AddedItem),
        }
    }
}

Each of the supported protocols has its benefits and drawbacks, and there probably isn't one that would cover all use cases.

The following table summarizes why each protocol was chosen:

While there are no benchmarks for timesource yet, it may be useful to check serdebench for serialisation benchmarks.

Difference with Eventually

Timesource was born out as a fork of eventually-rs. Eventually-rs doesn't support timestamps, and so it isn't possible to use timescaledb as storage engine in a meaningful way.

Whereas Eventually's philosophy is to be an event sourcing library with pluggable storage engines, Timesource will only ever support Timescaledb. This makes Timesource much narrower in scope. It also allows for a number of optimisations, a more efficient handling of events and the removal of the risk for memory overflows.

How to use

See the Orders aggregate example