haematite 0.7.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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110

//! 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 crate::db::root_advance::RootTransition;

use super::handle::{
    AdvanceReply, MergeAdoptReply, RangeItem, ScanReply, ShardCommand, ShardCommandKind,
    ShardError, ShardSyncExport,
};
use super::{PromiseState, RecordPromiseOutcome};

/// 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, .. } | ShardCommandKind::GetRaw { reply, .. } => {
            send_get(&reply, error);
        }
        ShardCommandKind::DeleteIfExpired { reply, .. }
        | ShardCommandKind::HasLiveInRange(_, _, reply) => send_bool(&reply, error),
        ShardCommandKind::Put { reply, .. }
        | ShardCommandKind::Delete { reply, .. }
        | ShardCommandKind::RecordOwnerEpoch { reply, .. }
        | ShardCommandKind::Shutdown { reply } => send_unit(&reply, error),
        ShardCommandKind::Cas { reply, .. }
        | ShardCommandKind::ApplyDurable { reply, .. }
        | ShardCommandKind::ApplyDurableTombstone { reply, .. }
        | ShardCommandKind::ApplyDurableBatch { reply, .. } => send_advance(&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::RecordPromise { reply, .. } => send_promise(&reply, error),
        ShardCommandKind::ReserveMinted { reply, .. } => send_reserved(&reply, error),
        ShardCommandKind::ReadPromiseState { reply } => send_promise_state(&reply, error),
        ShardCommandKind::ExportReachable { reply, .. } => send_export(&reply, error),
        ShardCommandKind::MergeAdopt { reply, .. } => send_merge_adopt(&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_advance(reply: &AdvanceReply, error: ShardError) {
    drop(reply.send(Err(error)));
}

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

fn send_commit(
    reply: &SyncSender<Result<(Hash, Option<RootTransition>), 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, Option<RootTransition>), 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)));
}

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

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

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

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

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