///|
/// The WAL engine.
///
/// One compare-and-swap on the image object is the only thing that makes a
/// commit visible. There is no lease, no election and no quorum: the store's
/// conditional write is the whole coordination mechanism.
///|
pub struct Wal[S] {
store : S
/// Key prefix for this log, e.g. `repos/acme/app`.
prefix : String
/// Last image read, and the version tag it was read at. The tag is empty
/// when the image has not been read, or is known not to exist.
mut image : WalImage
mut etag : String
/// Whether `image`/`etag` reflect a completed read.
mut loaded : Bool
max_attempts : Int
max_image_bytes : Int
}
///|
pub fn[S] Wal::new(
store : S,
prefix : String,
max_attempts? : Int = DEFAULT_MAX_ATTEMPTS,
max_image_bytes? : Int = DEFAULT_MAX_IMAGE_BYTES,
) -> Wal[S] {
{
store,
prefix,
image: WalImage::empty(),
etag: "",
loaded: false,
max_attempts,
max_image_bytes,
}
}
///|
pub fn[S] Wal::image_key(self : Wal[S]) -> String {
"\{self.prefix}/wal.json"
}
///|
pub fn[S] Wal::payload_key(self : Wal[S], content_id : String) -> String {
"\{self.prefix}/payloads/\{content_id}"
}
///|
pub fn[S] Wal::snapshot_key(self : Wal[S], content_id : String) -> String {
"\{self.prefix}/snapshots/\{content_id}"
}
///|
/// Content address for a blob: the same bytes always get the same key, which
/// is what makes writing one idempotent.
pub fn content_id(body : Bytes) -> String {
@hash.hex_encode(@hash.sha256_raw(body))
}
///|
/// Refresh the cached image.
///
/// After the first read this is a conditional GET: an unchanged log costs one
/// request that transfers nothing, which is what makes a replica cheap to
/// keep fresh.
pub async fn[S : @objstore.ObjectStore] Wal::sync(
self : Wal[S],
) -> Unit raise @bit.GitError {
let key = self.image_key()
let outcome = if self.loaded && self.etag != "" {
self.store.get_if_none_match(key, self.etag)
} else {
self.store.get(key)
}
match outcome {
NotModified => ()
Found(body, etag) => {
self.image = decode_image(body)
self.etag = etag
self.loaded = true
}
Missing => {
// Either the log has never been written, or it was deleted underneath
// us. Both mean "there is nothing committed", and a create-only write
// is the correct next move.
self.image = WalImage::empty()
self.etag = ""
self.loaded = true
}
}
}
///|
/// The live entries as of the last `sync`.
pub fn[S] Wal::entries(self : Wal[S]) -> Array[WalEntry] {
self.image.entries
}
///|
/// Sequence of the newest entry as of the last `sync`.
pub fn[S] Wal::head_seq(self : Wal[S]) -> Int64 {
self.image.head_seq()
}
///|
/// Snapshot key as of the last `sync`, empty when nothing has been folded.
pub fn[S] Wal::snapshot(self : Wal[S]) -> String {
self.image.snapshot
}
///|
pub fn[S] Wal::settings(self : Wal[S]) -> String {
self.image.settings
}
///|
/// Store a payload under its content address.
///
/// Safe to repeat: the key is derived from the bytes, so a repeat writes what
/// is already there. Returns the content id to reference from an entry.
pub async fn[S : @objstore.ObjectStore] Wal::put_payload(
self : Wal[S],
body : Bytes,
) -> String raise @bit.GitError {
let id = content_id(body)
let outcome = @objstore.put_immutable(self.store, self.payload_key(id), body)
match outcome {
Applied(_) => id
NotApplied =>
// put_immutable already folds the lost-race case into Applied, so this
// is a precondition failure that should not be reachable.
raise @bit.GitError::IoError("payload write refused for \{id}")
Unknown =>
raise @bit.GitError::IoError(
"payload write for \{id} is indeterminate; retry before committing",
)
}
}
///|
pub async fn[S : @objstore.ObjectStore] Wal::get_payload(
self : Wal[S],
content_id : String,
) -> Bytes raise @bit.GitError {
let key = self.payload_key(content_id)
let (body, _) = self.store.get(key).unwrap(key)
body
}
///|
/// Whether every one of these idempotency keys is already in the image.
///
/// This is how an indeterminate write is resolved: rather than guessing
/// whether the request landed, the writer looks for its own mark.
fn already_committed(image : WalImage, keys : Array[String]) -> Bool {
if keys.length() == 0 {
return false
}
let present : Map[String, Bool] = Map([])
for entry in image.entries {
if entry.idempotency_key != "" {
present[entry.idempotency_key] = true
}
}
for key in keys {
if key == "" || !present.contains(key) {
return false
}
}
true
}
///|
fn idempotency_keys(entries : Array[WalEntry]) -> Array[String] {
let out : Array[String] = []
for entry in entries {
out.push(entry.idempotency_key)
}
out
}
///|
/// Append entries as one atomic commit.
///
/// Several entries in one call are a group commit: they cost a single
/// compare-and-swap between them, which is what keeps concurrent pushes from
/// each paying their own round trip.
///
/// Entries carry payload keys but not payload bytes — write those with
/// `put_payload` first. That ordering is deliberate: a payload is content
/// addressed and immutable, so writing one that is never referenced leaves
/// garbage, while committing an entry that references a payload not yet
/// written would leave the log describing something that does not exist.
pub async fn[S : @objstore.ObjectStore] Wal::append(
self : Wal[S],
entries : Array[WalEntry],
) -> CommitOutcome raise @bit.GitError {
if entries.length() == 0 {
self.sync()
return CommitOutcome::Committed(self.head_seq())
}
if !self.store.supports_cas() {
raise @bit.GitError::IoError(
"backing store does not support conditional writes; " +
"a log cannot be hosted on it safely",
)
}
let keys = idempotency_keys(entries)
for _ in 0.. self.max_image_bytes {
raise @bit.GitError::IoError(
"WAL image would be \{body.length()} bytes, over the " +
"\{self.max_image_bytes} byte limit; compact before appending",
)
}
let condition = if self.etag == "" {
@objstore.PutCondition::IfNotExists
} else {
@objstore.PutCondition::IfMatch(self.etag)
}
match self.store.put(self.image_key(), body, condition) {
Applied(new_etag) => {
self.image = next_image
self.etag = new_etag
self.loaded = true
return CommitOutcome::Committed(seq)
}
// Someone else committed first. Re-read and rebuild on top of them.
NotApplied => self.invalidate()
// The write may have landed. The next pass re-reads and looks for our
// idempotency keys, so this neither double-applies nor gives up.
Unknown => self.invalidate()
}
}
CommitOutcome::Contended
}
///|
/// Drop the cached image so the next `sync` performs a full read.
fn[S] Wal::invalidate(self : Wal[S]) -> Unit {
self.loaded = false
self.etag = ""
}
///|
/// Replace the settings blob, under the same compare-and-swap discipline as
/// an append.
pub async fn[S : @objstore.ObjectStore] Wal::set_settings(
self : Wal[S],
settings : String,
) -> CommitOutcome raise @bit.GitError {
for _ in 0.. {
self.image = next_image
self.etag = new_etag
self.loaded = true
return CommitOutcome::Committed(next_image.head_seq())
}
NotApplied => self.invalidate()
Unknown => self.invalidate()
}
}
CommitOutcome::Contended
}