haematite 0.1.0

Content-addressed, branchable, actor-native storage engine
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

//! Boot-failure reply fan-out for a shard native process.
//!
//! When a shard fails to boot (bad store/WAL path, failed recovery) its
//! sentinel slice drains every queued command with a [`ShardError::Spawn`] so
//! callers fail fast instead of hanging. Each command kind carries a distinctly
//! typed reply channel, so the fan-out needs one tiny sender per kind.

use std::sync::mpsc::SyncSender;

use crate::tree::Hash;

use super::handle::{RangeItem, ScanReply, ShardCommand, ShardCommandKind, ShardError};

/// Reply to a queued command with a startup error so its caller fails fast.
pub(super) fn reply_startup_error(command: ShardCommand, message: &str) {
    let error = ShardError::Spawn(message.to_owned());
    match command.kind {
        ShardCommandKind::Get { reply, .. } => send_get(&reply, error),
        ShardCommandKind::Put { reply, .. }
        | ShardCommandKind::Delete { reply, .. }
        | ShardCommandKind::Cas { reply, .. }
        | ShardCommandKind::Shutdown { reply } => send_unit(&reply, error),
        ShardCommandKind::Commit { reply } => send_commit(&reply, error),
        ShardCommandKind::Range { reply, .. } => send_range(&reply, error),
        ShardCommandKind::Append { reply, .. } => send_append(&reply, error),
        ShardCommandKind::ReadValue { reply, .. } => send_read_value(&reply, error),
        ShardCommandKind::ScanSequences { reply } => send_scan(&reply, error),
    }
}

fn send_get(reply: &SyncSender<Result<Option<Vec<u8>>, ShardError>>, error: ShardError) {
    drop(reply.send(Err(error)));
}

fn send_unit(reply: &SyncSender<Result<(), ShardError>>, error: ShardError) {
    drop(reply.send(Err(error)));
}

fn send_commit(reply: &SyncSender<Result<Hash, ShardError>>, error: ShardError) {
    drop(reply.send(Err(error)));
}

fn send_range(reply: &SyncSender<Result<Vec<RangeItem>, ShardError>>, error: ShardError) {
    drop(reply.send(Err(error)));
}

fn send_append(reply: &SyncSender<Result<u64, ShardError>>, error: ShardError) {
    drop(reply.send(Err(error)));
}

fn send_read_value(reply: &SyncSender<Result<Option<u64>, ShardError>>, error: ShardError) {
    drop(reply.send(Err(error)));
}

fn send_scan(reply: &ScanReply, error: ShardError) {
    drop(reply.send(Err(error)));
}