use core::fmt::Write as _;
use std::sync::atomic::Ordering::Relaxed;
use std::sync::atomic::{AtomicBool, AtomicU64};
use yo_common::lock::Lock;
use yo_common::{Code, Error, Result};
use crate::reply::Out;
use super::args::{self, Args};
use super::{Server, Session};
mod asm;
mod bus;
use super::keyspec;
use super::table::Spec;
pub(super) use asm::Migration;
pub const SLOTS: usize = 16384;
const WRITABLE_DELAY_MS: u64 = 2000;
const REJOIN_DELAY_MS: u64 = 5000;
const BUS_OFFSET: u16 = 10000;
const ID_LEN: usize = 40;
#[rustfmt::skip]
const CRC16: [u16; 256] = [
0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50a5, 0x60c6, 0x70e7,
0x8108, 0x9129, 0xa14a, 0xb16b, 0xc18c, 0xd1ad, 0xe1ce, 0xf1ef,
0x1231, 0x0210, 0x3273, 0x2252, 0x52b5, 0x4294, 0x72f7, 0x62d6,
0x9339, 0x8318, 0xb37b, 0xa35a, 0xd3bd, 0xc39c, 0xf3ff, 0xe3de,
0x2462, 0x3443, 0x0420, 0x1401, 0x64e6, 0x74c7, 0x44a4, 0x5485,
0xa56a, 0xb54b, 0x8528, 0x9509, 0xe5ee, 0xf5cf, 0xc5ac, 0xd58d,
0x3653, 0x2672, 0x1611, 0x0630, 0x76d7, 0x66f6, 0x5695, 0x46b4,
0xb75b, 0xa77a, 0x9719, 0x8738, 0xf7df, 0xe7fe, 0xd79d, 0xc7bc,
0x48c4, 0x58e5, 0x6886, 0x78a7, 0x0840, 0x1861, 0x2802, 0x3823,
0xc9cc, 0xd9ed, 0xe98e, 0xf9af, 0x8948, 0x9969, 0xa90a, 0xb92b,
0x5af5, 0x4ad4, 0x7ab7, 0x6a96, 0x1a71, 0x0a50, 0x3a33, 0x2a12,
0xdbfd, 0xcbdc, 0xfbbf, 0xeb9e, 0x9b79, 0x8b58, 0xbb3b, 0xab1a,
0x6ca6, 0x7c87, 0x4ce4, 0x5cc5, 0x2c22, 0x3c03, 0x0c60, 0x1c41,
0xedae, 0xfd8f, 0xcdec, 0xddcd, 0xad2a, 0xbd0b, 0x8d68, 0x9d49,
0x7e97, 0x6eb6, 0x5ed5, 0x4ef4, 0x3e13, 0x2e32, 0x1e51, 0x0e70,
0xff9f, 0xefbe, 0xdfdd, 0xcffc, 0xbf1b, 0xaf3a, 0x9f59, 0x8f78,
0x9188, 0x81a9, 0xb1ca, 0xa1eb, 0xd10c, 0xc12d, 0xf14e, 0xe16f,
0x1080, 0x00a1, 0x30c2, 0x20e3, 0x5004, 0x4025, 0x7046, 0x6067,
0x83b9, 0x9398, 0xa3fb, 0xb3da, 0xc33d, 0xd31c, 0xe37f, 0xf35e,
0x02b1, 0x1290, 0x22f3, 0x32d2, 0x4235, 0x5214, 0x6277, 0x7256,
0xb5ea, 0xa5cb, 0x95a8, 0x8589, 0xf56e, 0xe54f, 0xd52c, 0xc50d,
0x34e2, 0x24c3, 0x14a0, 0x0481, 0x7466, 0x6447, 0x5424, 0x4405,
0xa7db, 0xb7fa, 0x8799, 0x97b8, 0xe75f, 0xf77e, 0xc71d, 0xd73c,
0x26d3, 0x36f2, 0x0691, 0x16b0, 0x6657, 0x7676, 0x4615, 0x5634,
0xd94c, 0xc96d, 0xf90e, 0xe92f, 0x99c8, 0x89e9, 0xb98a, 0xa9ab,
0x5844, 0x4865, 0x7806, 0x6827, 0x18c0, 0x08e1, 0x3882, 0x28a3,
0xcb7d, 0xdb5c, 0xeb3f, 0xfb1e, 0x8bf9, 0x9bd8, 0xabbb, 0xbb9a,
0x4a75, 0x5a54, 0x6a37, 0x7a16, 0x0af1, 0x1ad0, 0x2ab3, 0x3a92,
0xfd2e, 0xed0f, 0xdd6c, 0xcd4d, 0xbdaa, 0xad8b, 0x9de8, 0x8dc9,
0x7c26, 0x6c07, 0x5c64, 0x4c45, 0x3ca2, 0x2c83, 0x1ce0, 0x0cc1,
0xef1f, 0xff3e, 0xcf5d, 0xdf7c, 0xaf9b, 0xbfba, 0x8fd9, 0x9ff8,
0x6e17, 0x7e36, 0x4e55, 0x5e74, 0x2e93, 0x3eb2, 0x0ed1, 0x1ef0,
];
#[must_use]
fn crc16(data: &[u8]) -> u16 {
let mut crc: u16 = 0;
for &byte in data {
let at = ((crc >> 8) ^ u16::from(byte)) & 0xff;
crc = (crc << 8) ^ CRC16[at as usize];
}
crc
}
#[must_use]
pub fn key_slot(key: &[u8]) -> u16 {
let tagged = match key.iter().position(|&b| b == b'{') {
Some(open) => match key[open + 1..].iter().position(|&b| b == b'}') {
Some(0) | None => key,
Some(len) => &key[open + 1..open + 1 + len],
},
None => key,
};
crc16(tagged) % SLOTS as u16
}
pub(crate) const FLAG_MASTER: u16 = 1;
pub(crate) const FLAG_SLAVE: u16 = 2;
pub(crate) const FLAG_PFAIL: u16 = 4;
pub(crate) const FLAG_FAIL: u16 = 8;
pub(crate) const FLAG_MYSELF: u16 = 16;
pub(crate) const FLAG_HANDSHAKE: u16 = 32;
pub(crate) const FLAG_NOADDR: u16 = 64;
pub(crate) const FLAG_MEET: u16 = 128;
pub(crate) const FLAG_MIGRATE_TO: u16 = 256;
pub(crate) const FLAG_NOFAILOVER: u16 = 512;
pub(crate) const FLAG_EXTENSIONS: u16 = 1024;
#[derive(Clone)]
struct Node {
id: String,
host: String,
port: u16,
bus: u16,
shard: String,
epoch: u64,
flags: u16,
master: Option<u16>,
ping_sent: u64,
pong_recv: u64,
data_recv: u64,
fail_time: u64,
offset: u64,
linked: bool,
reports: Vec<(String, u64)>,
}
impl Node {
fn new(id: String, host: String, port: u16, bus: u16, flags: u16, now: u64) -> Node {
Node {
id,
host,
port,
bus,
shard: String::from_utf8_lossy(&new_id()).into_owned(),
epoch: 0,
flags,
master: None,
ping_sent: 0,
pong_recv: 0,
data_recv: now,
fail_time: 0,
offset: 0,
linked: false,
reports: Vec::new(),
}
}
fn is_master(&self) -> bool {
self.flags & FLAG_SLAVE == 0
}
fn down(&self) -> bool {
self.flags & (FLAG_PFAIL | FLAG_FAIL) != 0
}
fn flag_names(&self, into: &mut String) {
const NAMES: [(u16, &str); 8] = [
(FLAG_MYSELF, "myself"),
(FLAG_MASTER, "master"),
(FLAG_SLAVE, "slave"),
(FLAG_PFAIL, "fail?"),
(FLAG_FAIL, "fail"),
(FLAG_HANDSHAKE, "handshake"),
(FLAG_NOADDR, "noaddr"),
(FLAG_NOFAILOVER, "nofailover"),
];
let mut first = true;
for (bit, name) in NAMES {
if self.flags & bit == 0 {
continue;
}
if !first {
into.push(',');
}
into.push_str(name);
first = false;
}
if first {
into.push_str("noflags");
}
}
fn address(&self, into: &mut String) {
let _ = write!(into, "{}:{}@{}", self.host, self.port, self.bus);
}
fn address_on_disk(&self, into: &mut String) {
self.address(into);
let _ = write!(into, ",,tls-port=0,shard-id={}", self.shard);
}
}
struct Map {
nodes: Vec<Node>,
owner: Vec<Option<u16>>,
migrating: Vec<Option<u16>>,
importing: Vec<Option<u16>>,
}
impl Map {
fn new(me: Node) -> Map {
Map {
nodes: vec![me],
owner: vec![None; SLOTS],
migrating: vec![None; SLOTS],
importing: vec![None; SLOTS],
}
}
fn find(&self, id: &[u8]) -> Option<u16> {
self.nodes
.iter()
.position(|n| n.id.as_bytes() == id)
.map(|at| at as u16)
}
fn forget(&mut self, at: u16) {
self.nodes.remove(usize::from(at));
let shift = |slot: &mut Option<u16>| match *slot {
Some(node) if node == at => *slot = None,
Some(node) if node > at => *slot = Some(node - 1),
_ => {}
};
for slot in 0..SLOTS {
shift(&mut self.owner[slot]);
shift(&mut self.migrating[slot]);
shift(&mut self.importing[slot]);
}
for node in &mut self.nodes {
shift(&mut node.master);
}
}
fn voters(&self) -> usize {
let mut seen = vec![false; self.nodes.len()];
for owner in self.owner.iter().flatten() {
seen[*owner as usize] = true;
}
seen.iter().filter(|s| **s).count()
}
fn mine(&self, slot: u16) -> bool {
self.owner[slot as usize] == Some(0)
}
fn assigned(&self) -> usize {
self.owner.iter().filter(|o| o.is_some()).count()
}
fn size(&self) -> usize {
let mut seen = vec![false; self.nodes.len()];
for owner in self.owner.iter().flatten() {
seen[*owner as usize] = true;
}
seen.iter().filter(|s| **s).count()
}
fn runs(&self, node: u16) -> Vec<(u16, u16)> {
let mut runs: Vec<(u16, u16)> = Vec::new();
for slot in 0..SLOTS as u16 {
if self.owner[slot as usize] != Some(node) {
continue;
}
match runs.last_mut() {
Some(last) if last.1 + 1 == slot => last.1 = slot,
_ => runs.push((slot, slot)),
}
}
runs
}
}
pub(crate) struct Cluster {
on: bool,
map: Lock<Map>,
epoch: AtomicU64,
covered_at: AtomicU64,
booted_at: AtomicU64,
was_down: AtomicBool,
full_coverage: AtomicBool,
reads_when_down: AtomicBool,
file: Lock<String>,
bus: bus::Bus,
asm: asm::Asm,
}
impl Default for Cluster {
fn default() -> Cluster {
Cluster {
on: false,
map: Lock::new(Map {
nodes: Vec::new(),
owner: Vec::new(),
migrating: Vec::new(),
importing: Vec::new(),
}),
epoch: AtomicU64::new(0),
covered_at: AtomicU64::new(0),
booted_at: AtomicU64::new(0),
was_down: AtomicBool::new(false),
full_coverage: AtomicBool::new(true),
reads_when_down: AtomicBool::new(false),
file: Lock::new(String::new()),
bus: bus::Bus::default(),
asm: asm::Asm::default(),
}
}
}
impl Server {
#[must_use]
pub fn cluster_enabled(&self) -> bool {
self.cluster.on
}
pub fn enable_cluster(&mut self, file: &str, port: u16) {
self.cluster.on = true;
self.cluster.booted_at.store(self.now_ms(), Relaxed);
let now = self.now_ms();
let me = yo_alloc::allow(|| {
Node::new(
String::from_utf8_lossy(&new_id()).into_owned(),
String::new(),
port,
port + BUS_OFFSET,
FLAG_MYSELF | FLAG_MASTER,
now,
)
});
let path = yo_alloc::allow(|| {
if file.is_empty() {
String::new()
} else {
self.dir().join(file).to_string_lossy().into_owned()
}
});
yo_alloc::allow(|| {
*self.cluster.map.lock() = Map::new(me);
*self.cluster.file.lock() = path;
*self.cluster.bus.secret.lock() = String::from_utf8_lossy(&new_id()).into_owned();
});
if let Err(e) = self.reload_cluster() {
eprintln!("cluster config file could not be read: {e}");
}
self.recount_coverage();
}
pub(crate) fn cluster_secret(&self) -> String {
let held = self.cluster.bus.secret.lock();
yo_alloc::allow(|| held.clone())
}
pub(crate) fn cluster_full_coverage(&self) -> bool {
self.cluster.full_coverage.load(Relaxed)
}
pub(crate) fn cluster_reads_when_down(&self) -> bool {
self.cluster.reads_when_down.load(Relaxed)
}
pub(crate) fn set_cluster_coverage(&self, full: bool, reads_when_down: bool) {
self.cluster.full_coverage.store(full, Relaxed);
self.cluster.reads_when_down.store(reads_when_down, Relaxed);
self.recount_coverage();
}
pub(crate) fn cluster_file(&self) -> String {
let file = self.cluster.file.lock();
yo_alloc::allow(|| file.clone())
}
pub(crate) fn cluster_id(&self) -> String {
let map = self.cluster.map.lock();
yo_alloc::allow(|| map.nodes.first().map_or_else(String::new, |n| n.id.clone()))
}
pub(crate) fn cluster_up(&self) -> bool {
let at = self.cluster.covered_at.load(Relaxed);
if at == 0 {
return false;
}
let (since, wait) = if self.cluster.was_down.load(Relaxed) {
(at, REJOIN_DELAY_MS)
} else {
(self.cluster.booted_at.load(Relaxed), WRITABLE_DELAY_MS)
};
self.now_ms().saturating_sub(since) >= wait
}
fn recount_coverage(&self) {
let covered = {
let map = self.cluster.map.lock();
let assigned = map.assigned();
if self.cluster_full_coverage() {
assigned == SLOTS
} else {
assigned > 0
}
};
if covered {
let _ =
self.cluster
.covered_at
.compare_exchange(0, self.now_ms().max(1), Relaxed, Relaxed);
} else {
self.cluster.covered_at.store(0, Relaxed);
self.cluster.was_down.store(true, Relaxed);
}
}
}
fn new_id() -> [u8; ID_LEN] {
const HEX: &[u8; 16] = b"0123456789abcdef";
let mut raw = [0u8; ID_LEN / 2];
yo_common::entropy::fill(&mut raw);
let mut id = [0u8; ID_LEN];
for (i, byte) in raw.iter().enumerate() {
id[i * 2] = HEX[usize::from(byte >> 4)];
id[i * 2 + 1] = HEX[usize::from(byte & 15)];
}
id
}
pub(super) fn asks(spec: &Spec) -> bool {
spec.flags.contains(&"asking")
}
pub(super) fn gate(
server: &Server,
db: usize,
asking: bool,
spec: &Spec,
args: Args<'_>,
) -> Option<Error> {
if !keyspec::takes_keys(spec, args, 0) {
return None;
}
let mut slot: Option<u16> = None;
let mut crossed = false;
let mut keys = 0usize;
let mut present = 0usize;
let mut missing = 0usize;
let (owner, migrating, importing, here) = {
let map = server.cluster.map.lock();
keyspec::find(spec, args, 0, &mut |run| {
for i in 0..run.count {
let at = run.first + i * run.step;
if at >= args.len() {
continue;
}
let this = key_slot(args.get(at));
keys += 1;
match slot {
None => slot = Some(this),
Some(first) if first != this => crossed = true,
Some(_) => {}
}
}
});
let at = usize::from(slot?);
(
map.owner[at],
map.migrating[at].map(|to| node_at(&map, to)),
map.importing[at].is_some(),
map.owner[at] == Some(0),
)
};
let slot = slot?;
let Some(owner) = owner else {
return Some(Error::new(
Code::Invalid,
"CLUSTERDOWN Hash slot not served",
));
};
if crossed {
return Some(Error::new(
Code::Invalid,
"CROSSSLOT Keys in request don't hash to the same slot",
));
}
if !server.cluster_up() {
if !server.cluster.reads_when_down.load(Relaxed) {
return Some(Error::new(Code::Invalid, "CLUSTERDOWN The cluster is down"));
}
if spec.flags.contains(&"write") {
return Some(Error::new(
Code::Invalid,
"CLUSTERDOWN The cluster is down and only accepts read commands",
));
}
}
if migrating.is_some() || importing {
let held = &server.dbs[db];
keyspec::find(spec, args, 0, &mut |run| {
for i in 0..run.count {
let at = run.first + i * run.step;
if at >= args.len() {
continue;
}
let key = args.get(at);
let mut stripe = held.hold(key);
if stripe.exists(key) {
present += 1;
} else {
missing += 1;
}
}
});
}
if let Some(to) = migrating
&& missing > 0
{
if present > 0 {
return Some(Error::new(
Code::Invalid,
"TRYAGAIN Multiple keys request during rehashing of slot",
));
}
return Some(redirect("ASK", slot, &to));
}
if importing && asking {
if keys > 1 && missing > 0 {
return Some(Error::new(
Code::Invalid,
"TRYAGAIN Multiple keys request during rehashing of slot",
));
}
return None;
}
if here {
return None;
}
let (host, port) = {
let map = server.cluster.map.lock();
node_at(&map, owner)
};
Some(redirect("MOVED", slot, &(host, port)))
}
fn node_at(map: &Map, at: u16) -> (String, u16) {
let node = &map.nodes[at as usize];
(yo_alloc::allow(|| node.host.clone()), node.port)
}
fn redirect(word: &str, slot: u16, node: &(String, u16)) -> Error {
let host = if node.0.is_empty() {
"127.0.0.1"
} else {
node.0.as_str()
};
Error::fmt(
Code::Invalid,
format_args!("{word} {slot} {host}:{}", node.1),
)
}
pub(super) fn execute(
server: &Server,
session: &mut Session,
args: Args<'_>,
out: &mut Out,
) -> Result<()> {
let session_db = session.db;
let sub = args.get(1);
if !server.cluster_enabled() {
return match arity_of(sub) {
Some(n) if !arity_ok(n, args.len()) => Err(wrong_sub_arity(sub)),
Some(_) => Err(disabled()),
None => Err(args::unknown_subcommand(sub, "CLUSTER")),
};
}
let Some(n) = arity_of(sub) else {
return Err(args::unknown_subcommand(sub, "CLUSTER"));
};
if !arity_ok(n, args.len()) {
return Err(wrong_sub_arity(sub));
}
match () {
() if args::is(sub, b"myid") => out.bulk(server.cluster_id().as_bytes()),
() if args::is(sub, b"myshardid") => {
let map = server.cluster.map.lock();
out.bulk(map.nodes[0].shard.as_bytes());
}
() if args::is(sub, b"keyslot") => out.int(i64::from(key_slot(args.get(2)))),
() if args::is(sub, b"info") => info(server, out),
() if args::is(sub, b"nodes") => nodes(server, out),
() if args::is(sub, b"slots") => reply_slots(server, out),
() if args::is(sub, b"shards") => shards(server, out),
() if args::is(sub, b"links") => server.cluster_links(out),
() if args::is(sub, b"slaves") || args::is(sub, b"replicas") => {
replicas(server, args.get(2), out)?;
}
() if args::is(sub, b"count-failure-reports") => {
let map = server.cluster.map.lock();
let Some(at) = map.find(args.get(2)) else {
return Err(unknown_node(args.get(2)));
};
out.int(map.nodes[usize::from(at)].reports.len() as i64);
}
() if args::is(sub, b"countkeysinslot") => count_keys(server, session_db, args, out)?,
() if args::is(sub, b"getkeysinslot") => get_keys(server, session_db, args, out)?,
() if args::is(sub, b"addslots") => {
add_or_del(server, args, true, false)?;
out.ok();
}
() if args::is(sub, b"delslots") => {
add_or_del(server, args, false, false)?;
out.ok();
}
() if args::is(sub, b"addslotsrange") => {
add_or_del(server, args, true, true)?;
out.ok();
}
() if args::is(sub, b"delslotsrange") => {
add_or_del(server, args, false, true)?;
out.ok();
}
() if args::is(sub, b"setslot") => setslot(server, session_db, args, out)?,
() if args::is(sub, b"flushslots") => flushslots(server, out)?,
() if args::is(sub, b"bumpepoch") => bumpepoch(server, out)?,
() if args::is(sub, b"set-config-epoch") => set_config_epoch(server, args, out)?,
() if args::is(sub, b"reset") => {
if args.len() > 3 {
return Err(sub_syntax(sub));
}
reset(server, args, out)?;
}
() if args::is(sub, b"slot-stats") => slot_stats(server, session_db, args, out)?,
() if args::is(sub, b"migration") => migration(server, args, out)?,
() if args::is(sub, b"syncslots") => syncslots(server, session, args, out)?,
() if args::is(sub, b"saveconfig") => {
save(server)?;
out.ok();
}
() if args::is(sub, b"forget") => {
forget(server, args.get(2))?;
out.ok();
}
() if args::is(sub, b"replicate") => {
replicate(server, session_db, args.get(2))?;
out.ok();
}
() if args::is(sub, b"failover") => {
if args.len() > 3 {
return Err(sub_syntax(sub));
}
if args.len() == 3
&& !args::is(args.get(2), b"force")
&& !args::is(args.get(2), b"takeover")
{
return Err(args::syntax());
}
return Err(Error::new(
Code::Invalid,
"You should send CLUSTER FAILOVER to a replica",
));
}
() if args::is(sub, b"meet") => {
if args.len() > 5 {
return Err(sub_syntax(sub));
}
let (host, port, bus) = meet(&args)?;
server.cluster_meet(&host, port, bus);
out.ok();
}
() if args::is(sub, b"help") => help(out),
_ => return Err(args::unknown_subcommand(sub, "CLUSTER")),
}
Ok(())
}
pub(super) fn disabled() -> Error {
Error::new(Code::Invalid, "This instance has cluster support disabled")
}
fn not_yet(what: &str) -> Error {
Error::fmt(
Code::Invalid,
format_args!("{what} is not implemented yet, move the slot with SETSLOT and MIGRATE"),
)
}
fn unknown_node(id: &[u8]) -> Error {
Error::fmt(
Code::Invalid,
format_args!("Unknown node {}", String::from_utf8_lossy(id)),
)
}
fn dont_know(id: &[u8]) -> Error {
Error::fmt(
Code::Invalid,
format_args!("I don't know about node {}", String::from_utf8_lossy(id)),
)
}
fn arity_of(sub: &[u8]) -> Option<i32> {
const TABLE: &[(&str, i32)] = &[
("addslots", -3),
("addslotsrange", -4),
("bumpepoch", 2),
("count-failure-reports", 3),
("countkeysinslot", 3),
("delslots", -3),
("delslotsrange", -4),
("failover", -2),
("flushslots", 2),
("forget", 3),
("getkeysinslot", 4),
("help", 2),
("info", 2),
("keyslot", 3),
("links", 2),
("meet", -4),
("migration", -4),
("myid", 2),
("myshardid", 2),
("nodes", 2),
("replicas", 3),
("replicate", 3),
("reset", -2),
("saveconfig", 2),
("set-config-epoch", 3),
("setslot", -4),
("shards", 2),
("slaves", 3),
("slot-stats", -4),
("slots", 2),
("syncslots", -3),
];
TABLE
.iter()
.find(|(name, _)| args::is(sub, name.as_bytes()))
.map(|(_, arity)| *arity)
}
fn arity_ok(arity: i32, len: usize) -> bool {
let len = len as i32;
if arity >= 0 {
len == arity
} else {
len >= -arity
}
}
fn sub_syntax(sub: &[u8]) -> Error {
Error::fmt(
Code::Unsupported,
format_args!(
"unknown subcommand or wrong number of arguments for '{}'. Try CLUSTER HELP.",
String::from_utf8_lossy(sub)
),
)
}
fn wrong_sub_arity(sub: &[u8]) -> Error {
Error::fmt(
Code::Invalid,
format_args!(
"wrong number of arguments for 'cluster|{}' command",
String::from_utf8_lossy(sub).to_lowercase()
),
)
}
fn info(server: &Server, out: &mut Out) {
let (assigned, size, known, my_epoch) = {
let map = server.cluster.map.lock();
(
map.assigned(),
map.size(),
map.nodes.len(),
map.nodes[0].epoch,
)
};
let state = if server.cluster_up() { "ok" } else { "fail" };
let text = yo_alloc::allow(|| {
let mut s = String::with_capacity(512);
let _ = write!(
s,
"cluster_state:{state}\r\ncluster_slots_assigned:{assigned}\r\n\
cluster_slots_ok:{assigned}\r\ncluster_slots_pfail:0\r\ncluster_slots_fail:0\r\n\
cluster_known_nodes:{known}\r\ncluster_size:{size}\r\n\
cluster_current_epoch:{}\r\ncluster_my_epoch:{my_epoch}\r\n\
cluster_stats_messages_sent:0\r\ncluster_stats_messages_received:0\r\n\
total_cluster_links_buffer_limit_exceeded:0\r\n\
cluster_slot_migration_active_tasks:0\r\n\
cluster_slot_migration_active_trim_running:0\r\n\
cluster_slot_migration_active_trim_current_job_keys:0\r\n\
cluster_slot_migration_active_trim_current_job_trimmed:0\r\n\
cluster_slot_migration_stats_active_trim_started:0\r\n\
cluster_slot_migration_stats_active_trim_completed:0\r\n\
cluster_slot_migration_stats_active_trim_cancelled:0\r\n",
server.cluster.epoch.load(Relaxed),
);
s
});
out.verbatim(b"txt", text.as_bytes());
}
fn nodes(server: &Server, out: &mut Out) {
let text = yo_alloc::allow(|| lines(server, false));
out.verbatim(b"txt", text.as_bytes());
}
fn lines(server: &Server, on_disk: bool) -> String {
let map = server.cluster.map.lock();
let mut s = String::with_capacity(256);
for at in 0..map.nodes.len() as u16 {
describe(&map, at, on_disk, &mut s);
s.push('\n');
}
s
}
fn describe(map: &Map, at: u16, on_disk: bool, s: &mut String) {
let node = &map.nodes[usize::from(at)];
s.push_str(&node.id);
s.push(' ');
if on_disk {
node.address_on_disk(s);
} else {
node.address(s);
}
s.push(' ');
node.flag_names(s);
s.push(' ');
match node.master.and_then(|m| map.nodes.get(usize::from(m))) {
Some(master) => s.push_str(&master.id),
None => s.push('-'),
}
let epoch = match node.master.and_then(|m| map.nodes.get(usize::from(m))) {
Some(master) => master.epoch,
None => node.epoch,
};
let link = if node.linked || at == 0 {
"connected"
} else {
"disconnected"
};
let _ = write!(s, " {} {} {epoch} {link}", node.ping_sent, node.pong_recv);
for (from, to) in map.runs(at) {
if from == to {
let _ = write!(s, " {from}");
} else {
let _ = write!(s, " {from}-{to}");
}
}
if at == 0 {
for slot in 0..SLOTS {
if let Some(to) = map.migrating[slot] {
let _ = write!(s, " [{slot}->-{}]", map.nodes[to as usize].id);
}
if let Some(from) = map.importing[slot] {
let _ = write!(s, " [{slot}-<-{}]", map.nodes[from as usize].id);
}
}
}
}
fn reply_slots(server: &Server, out: &mut Out) {
let mine = server.repl_offset();
let map = server.cluster.map.lock();
let at = out.len();
let mut n = 0;
let mut run: Option<(u16, u16)> = None;
for slot in 0..=SLOTS as u16 {
let owner = if slot as usize == SLOTS {
None
} else {
map.owner[slot as usize]
};
match run {
Some((node, _)) if owner == Some(node) => {}
Some((node, from)) => {
slot_run(&map, node, from, slot - 1, mine, out);
n += 1;
run = owner.map(|node| (node, slot));
}
None => run = owner.map(|node| (node, slot)),
}
}
out.close_array(at, n);
}
fn slot_run(map: &Map, node: u16, from: u16, to: u16, mine: u64, out: &mut Out) {
let replicas: Vec<&Node> = map
.nodes
.iter()
.enumerate()
.filter(|(at, n)| {
let offset = if *at == 0 { mine } else { n.offset };
n.master == Some(node) && n.flags & FLAG_FAIL == 0 && offset != 0
})
.map(|(_, n)| n)
.collect();
out.array(3 + replicas.len());
out.int(i64::from(from));
out.int(i64::from(to));
let held = &map.nodes[node as usize];
for held in std::iter::once(held).chain(replicas.iter().copied()) {
out.array(4);
out.bulk(held.host.as_bytes());
out.int(i64::from(held.port));
out.bulk(held.id.as_bytes());
out.array(0);
}
}
fn shards(server: &Server, out: &mut Out) {
let map = server.cluster.map.lock();
let at = out.len();
let mut n = 0;
let mut done: Vec<&str> = Vec::new();
for node in 0..map.nodes.len() {
let shard = map.nodes[node].shard.as_str();
if done.contains(&shard) {
continue;
}
done.push(shard);
let members: Vec<usize> = (0..map.nodes.len())
.filter(|other| map.nodes[*other].shard == shard)
.collect();
let runs: Vec<(u16, u16)> = members
.iter()
.flat_map(|member| map.runs(*member as u16))
.collect();
out.map(2);
out.bulk(b"slots");
out.array(runs.len() * 2);
for (from, to) in runs {
out.int(i64::from(from));
out.int(i64::from(to));
}
out.bulk(b"nodes");
out.array(members.len());
for member in members {
let held = &map.nodes[member];
out.map(7);
out.bulk(b"id");
out.bulk(held.id.as_bytes());
out.bulk(b"port");
out.int(i64::from(held.port));
out.bulk(b"ip");
out.bulk(held.host.as_bytes());
out.bulk(b"endpoint");
out.bulk(held.host.as_bytes());
out.bulk(b"role");
out.bulk(if held.is_master() {
b"master".as_slice()
} else {
b"replica".as_slice()
});
out.bulk(b"replication-offset");
out.int(if member == 0 {
server.repl_offset() as i64
} else {
held.offset as i64
});
out.bulk(b"health");
out.bulk(match held.flags {
f if f & FLAG_FAIL != 0 => b"fail".as_slice(),
f if f & FLAG_PFAIL != 0 => b"loading".as_slice(),
_ => b"online".as_slice(),
});
}
n += 1;
}
out.close_array(at, n);
}
fn help(out: &mut Out) {
const LINES: &[&str] = &[
"CLUSTER <subcommand> [<arg> [value] [opt] ...]. Subcommands are:",
"COUNTKEYSINSLOT <slot>",
" Return the number of keys in <slot>.",
"GETKEYSINSLOT <slot> <count>",
" Return key names stored by current node in a slot.",
"INFO",
" Return information about the cluster.",
"KEYSLOT <key>",
" Return the hash slot for <key>.",
"MYID",
" Return the node id.",
"MYSHARDID",
" Return the node's shard id.",
"NODES",
" Return cluster configuration seen by node. Output format:",
" <id> <ip:port@bus-port[,hostname]> <flags> <master> <pings> <pongs> <epoch> <link> <slot> ...",
"REPLICAS <node-id>",
" Return <node-id> replicas.",
"SLOTS",
" Return information about slots range mappings. Each range is made of:",
" start, end, master and replicas IP addresses, ports and ids",
"SLOT-STATS",
" Return an array of slot usage statistics for slots assigned to the current node.",
"SHARDS",
" Return information about slot range mappings and the nodes associated with them.",
"ADDSLOTS <slot> [<slot> ...]",
" Assign slots to current node.",
"ADDSLOTSRANGE <start slot> <end slot> [<start slot> <end slot> ...]",
" Assign slots which are between <start-slot> and <end-slot> to current node.",
"BUMPEPOCH",
" Advance the cluster config epoch.",
"COUNT-FAILURE-REPORTS <node-id>",
" Return number of failure reports for <node-id>.",
"DELSLOTS <slot> [<slot> ...]",
" Delete slots information from current node.",
"DELSLOTSRANGE <start slot> <end slot> [<start slot> <end slot> ...]",
" Delete slots information which are between <start-slot> and <end-slot> from current node.",
"FAILOVER [FORCE|TAKEOVER]",
" Promote current replica node to being a master.",
"FORGET <node-id>",
" Remove a node from the cluster.",
"FLUSHSLOTS",
" Delete current node own slots information.",
"MEET <ip> <port> [<bus-port>]",
" Connect nodes into a working cluster.",
"REPLICATE <node-id>",
" Configure current node as replica to <node-id>.",
"RESET [HARD|SOFT]",
" Reset current node (default: soft).",
"SET-CONFIG-EPOCH <epoch>",
" Set config epoch of current node.",
"SETSLOT <slot> (IMPORTING <node-id>|MIGRATING <node-id>|STABLE|NODE <node-id>)",
" Set slot state.",
"SAVECONFIG",
" Force saving cluster configuration on disk.",
"LINKS",
" Return information about all network links between this node and its peers.",
" Output format is an array where each array element is a map containing attributes of a link",
"MIGRATION IMPORT <start-slot end-slot [start-slot end-slot ...]> |",
" STATUS [ID <task-id> | ALL] | CANCEL [ID <task-id> | ALL]",
" Start, monitor and cancel slot migration.",
"HELP",
" Print this help.",
];
out.array(LINES.len());
for line in LINES {
out.simple(line.as_bytes());
}
}
fn slot_arg(args: &Args<'_>, at: usize) -> Result<u16> {
args.int(at)
.ok()
.and_then(|n| u16::try_from(n).ok())
.filter(|s| usize::from(*s) < SLOTS)
.ok_or_else(|| Error::new(Code::Invalid, "Invalid or out of range slot"))
}
fn add_or_del(server: &Server, args: Args<'_>, add: bool, ranged: bool) -> Result<()> {
let stride = if ranged { 2 } else { 1 };
if ranged && !(args.len() - 2).is_multiple_of(2) {
return Err(wrong_sub_arity(args.get(1)));
}
let mut wanted = Vec::new();
let mut at = 2;
while at < args.len() {
let from = slot_arg(&args, at)?;
let to = if ranged {
slot_arg(&args, at + 1)?
} else {
from
};
if from > to {
return Err(Error::fmt(
Code::Invalid,
format_args!("start slot number {from} is greater than end slot number {to}"),
));
}
for slot in from..=to {
wanted.push(slot);
}
at += stride;
}
{
let mut map = server.cluster.map.lock();
let mut seen = vec![false; SLOTS];
for slot in &wanted {
let slot = usize::from(*slot);
if seen[slot] {
return Err(Error::fmt(
Code::Invalid,
format_args!("Slot {slot} specified multiple times"),
));
}
seen[slot] = true;
let busy = map.owner[slot].is_some();
if add && busy {
return Err(Error::fmt(
Code::Invalid,
format_args!("Slot {slot} is already busy"),
));
}
if !add && !busy {
return Err(Error::fmt(
Code::Invalid,
format_args!("Slot {slot} is already unassigned"),
));
}
}
for slot in &wanted {
let slot = usize::from(*slot);
map.owner[slot] = if add { Some(0) } else { None };
map.migrating[slot] = None;
map.importing[slot] = None;
}
}
server.recount_coverage();
save(server)
}
fn meet(args: &Args<'_>) -> Result<(String, u16, u16)> {
let host = String::from_utf8_lossy(args.get(2));
let typed = String::from_utf8_lossy(args.get(3));
let port = args.int(3).map_err(|_| {
Error::fmt(
Code::Invalid,
format_args!("Invalid base port specified: {typed}"),
)
})?;
let bus = match args.opt(4) {
None => port + i64::from(BUS_OFFSET),
Some(word) => args.int(4).map_err(|_| {
Error::fmt(
Code::Invalid,
format_args!(
"Invalid bus port specified: {}",
String::from_utf8_lossy(word)
),
)
})?,
};
if !(1..=65535).contains(&port) || !(0..=65535).contains(&bus) {
return Err(Error::fmt(
Code::Invalid,
format_args!("Invalid node address specified: {host}:{typed}"),
));
}
let bus = if bus == 0 {
port + i64::from(BUS_OFFSET)
} else {
bus
};
let host = yo_alloc::allow(|| host.into_owned());
Ok((host, port as u16, bus as u16))
}
fn replicas(server: &Server, id: &[u8], out: &mut Out) -> Result<()> {
let map = server.cluster.map.lock();
let Some(at) = map.find(id) else {
return Err(unknown_node(id));
};
if !map.nodes[usize::from(at)].is_master() {
return Err(Error::new(
Code::Invalid,
"The specified node is not a master",
));
}
let start = out.len();
let mut n = 0;
for other in 0..map.nodes.len() as u16 {
if map.nodes[usize::from(other)].master != Some(at) {
continue;
}
let line = yo_alloc::allow(|| {
let mut s = String::with_capacity(256);
describe(&map, other, false, &mut s);
s
});
out.bulk(line.as_bytes());
n += 1;
}
out.close_array(start, n);
Ok(())
}
fn forget(server: &Server, id: &[u8]) -> Result<()> {
let at = {
let map = server.cluster.map.lock();
match map.find(id) {
Some(0) => {
return Err(Error::new(
Code::Invalid,
"I tried hard but I can't forget myself...",
));
}
Some(at) if map.nodes[0].master == Some(at) => {
return Err(Error::new(Code::Invalid, "Can't forget my master!"));
}
Some(at) => at,
None => {
let name = String::from_utf8_lossy(id);
if server.cluster_blacklisted(&name) {
return Ok(());
}
return Err(unknown_node(id));
}
}
};
server.cluster_forget(at);
Ok(())
}
fn replicate(server: &Server, db: usize, id: &[u8]) -> Result<()> {
let at = {
let map = server.cluster.map.lock();
match map.find(id) {
None => return Err(unknown_node(id)),
Some(0) => return Err(Error::new(Code::Invalid, "Can't replicate myself")),
Some(at) if !map.nodes[usize::from(at)].is_master() => {
return Err(Error::new(
Code::Invalid,
"I can only replicate a master, not a replica.",
));
}
Some(at) => {
if map.nodes[0].is_master()
&& (!map.runs(0).is_empty() || !server.dbs[db].is_empty())
{
return Err(Error::new(
Code::Invalid,
"To set a master the node must be empty and without assigned slots.",
));
}
at
}
}
};
let Some(shared) = server.myself() else {
return Err(Error::new(
Code::Invalid,
"CLUSTER REPLICATE is not available on an embedded server",
));
};
shared.cluster_replicate(at);
Ok(())
}
fn slot_stats(server: &Server, db: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
let sub = args.get(1);
let mut wanted: Vec<u16> = Vec::new();
let mut limit = SLOTS;
let mut ascending = false;
let ordered = args::is(args.get(2), b"orderby");
if args::is(args.get(2), b"slotsrange") {
if args.len() != 5 {
return Err(sub_syntax(sub));
}
let from = slot_arg(&args, 3)?;
let to = slot_arg(&args, 4)?;
if from > to {
return Err(Error::fmt(
Code::Invalid,
format_args!("Start slot number {from} is greater than end slot number {to}"),
));
}
wanted.extend(from..=to);
} else if ordered {
if !args::is(args.get(3), b"key-count") {
return Err(Error::new(
Code::Invalid,
"Unrecognized sort metric for ORDERBY.",
));
}
let bad_limit = || {
Error::new(
Code::Invalid,
"Limit has to lie in between 1 and 16384 (maximum number of slots).",
)
};
let mut at = 4;
while at < args.len() {
let word = args.get(at);
if args::is(word, b"limit") && at + 1 < args.len() {
let n = args.int(at + 1).map_err(|_| bad_limit())?;
if !(1..=SLOTS as i64).contains(&n) {
return Err(bad_limit());
}
limit = n as usize;
at += 2;
} else if args::is(word, b"asc") {
ascending = true;
at += 1;
} else if args::is(word, b"desc") {
at += 1;
} else {
return Err(args::syntax());
}
}
wanted.extend(0..SLOTS as u16);
} else {
return Err(sub_syntax(sub));
}
let mut counts = vec![0i64; SLOTS];
server.dbs[db].keys(|key| counts[key_slot(key) as usize] += 1);
let map = server.cluster.map.lock();
wanted.retain(|slot| map.mine(*slot));
drop(map);
if ordered {
if ascending {
wanted.sort_by_key(|slot| (counts[*slot as usize], *slot));
} else {
wanted.sort_by_key(|slot| (-counts[*slot as usize], *slot));
}
wanted.truncate(limit);
}
out.array(wanted.len());
for slot in wanted {
out.array(2);
out.int(i64::from(slot));
out.map(1);
out.bulk(b"key-count");
out.int(counts[slot as usize]);
}
Ok(())
}
fn migration(server: &Server, args: Args<'_>, out: &mut Out) -> Result<()> {
let sub = args.get(1);
let action = args.get(2);
if args::is(action, b"status") || args::is(action, b"cancel") {
let by_id = args::is(args.get(3), b"id");
if by_id && args.len() != 5 {
return Err(wrong_sub_arity(sub));
}
if !by_id && !args::is(args.get(3), b"all") {
return Err(Error::new(Code::Invalid, "unknown argument"));
}
if !by_id && args.len() != 4 {
return Err(wrong_sub_arity(sub));
}
let id = by_id.then(|| args.get(4));
if args::is(action, b"status") {
match id {
Some(id) => server.cluster.asm.report_one(id, out),
None => server.cluster.asm.report_all(out),
}
} else {
out.int(server.cluster.asm.cancel(id, server.now_ms() as i64));
server.asm_relax();
}
return Ok(());
}
if !args::is(action, b"import") {
return Err(Error::new(Code::Invalid, "unknown argument"));
}
let ranges = slot_ranges(&args, 3)?;
let map = server.cluster.map.lock();
if ranges
.iter()
.flat_map(|(from, to)| *from..=*to)
.all(|slot| map.mine(slot))
{
return Err(Error::new(
Code::Invalid,
"this node is already the owner of the slot range",
));
}
drop(map);
Err(not_yet("CLUSTER MIGRATION IMPORT"))
}
fn slot_ranges(args: &Args<'_>, from: usize) -> Result<Vec<(u16, u16)>> {
let count = args.len().saturating_sub(from);
if count < 2 || !count.is_multiple_of(2) {
return Err(wrong_sub_arity(args.get(1)));
}
if count / 2 >= SLOTS {
return Err(Error::fmt(
Code::Invalid,
format_args!("invalid number of slot ranges: {}", count / 2),
));
}
let mut ranges: Vec<(u16, u16)> = Vec::with_capacity(count / 2);
let mut at = from;
while at < args.len() {
ranges.push((slot_arg(args, at)?, slot_arg(args, at + 1)?));
at += 2;
}
ranges.sort_unstable();
let mut joined: Vec<(u16, u16)> = Vec::with_capacity(ranges.len());
for range in ranges {
match joined.last_mut() {
Some(last) if u32::from(last.1) + 1 == u32::from(range.0) => last.1 = range.1,
_ => joined.push(range),
}
}
let mut seen = vec![false; SLOTS];
for &(start, end) in &joined {
if start > end {
return Err(Error::fmt(
Code::Invalid,
format_args!("start slot number {start} is greater than end slot number {end}"),
));
}
for slot in start..=end {
if core::mem::replace(&mut seen[usize::from(slot)], true) {
return Err(Error::fmt(
Code::Invalid,
format_args!("Slot {slot} specified multiple times"),
));
}
}
}
Ok(joined)
}
fn syncslots(server: &Server, session: &mut Session, args: Args<'_>, out: &mut Out) -> Result<()> {
if !session.internal() {
session.hang_up();
return Err(Error::new(
Code::Invalid,
"CLUSTER SYNCSLOTS subcommands are only allowed for internal clients",
));
}
let action = args.get(2);
if !server.cluster.map.lock().nodes[0].is_master() {
if !session.serving_master() {
session.hang_up();
return Err(Error::new(
Code::Invalid,
"CLUSTER SYNCSLOTS subcommands are only allowed for master",
));
}
if !args::is(action, b"conf") {
return Ok(());
}
}
if args::is(action, b"sync") && args.len() >= 6 {
return sync(server, session, args, out);
}
if args::is(action, b"rdbchannel") && args.len() == 4 {
return rdbchannel(server, session, args, out);
}
if (args::is(action, b"snapshot-eof") || args::is(action, b"stream-eof")) && args.len() == 3 {
session.hang_up();
return Ok(());
}
if args::is(action, b"ack") && args.len() == 5 {
if let Some(offset) = yo_common::num::parse_i64(args.get(4))
&& offset >= 0
{
server.asm_ack(session.row().id, args.get(3), offset as u64);
}
return Ok(());
}
if args::is(action, b"fail") && args.len() == 4 {
return Ok(());
}
if args::is(action, b"conf") && args.len() >= 5 {
return conf(server, session, args, out);
}
Err(args::syntax())
}
fn sync(server: &Server, session: &mut Session, args: Args<'_>, out: &mut Out) -> Result<()> {
if !args.len().is_multiple_of(2) {
return Err(wrong_sub_arity(args.get(1)));
}
let ranges = slot_ranges(&args, 4)?;
{
let map = server.cluster.map.lock();
if (0..SLOTS).any(|at| map.migrating[at].is_some() || map.importing[at].is_some()) {
return Err(Error::new(
Code::Invalid,
"all slot states must be STABLE to start a slot migration task.",
));
}
let mut source = None;
for slot in ranges.iter().flat_map(|(from, to)| *from..=*to) {
let Some(owner) = map.owner[usize::from(slot)] else {
return Err(Error::fmt(
Code::Invalid,
format_args!("slot has no owner: {slot}"),
));
};
if *source.get_or_insert(owner) != owner {
return Err(Error::new(
Code::Invalid,
"slots belong to different source nodes",
));
}
}
if source != Some(0) {
return Err(Error::new(
Code::Invalid,
"This node is not the owner of the slots",
));
}
let dest = session.node_id().to_vec();
if !dest.is_empty()
&& !map
.find(&dest)
.is_some_and(|at| map.nodes[at as usize].is_master())
{
return Err(Error::fmt(
Code::Invalid,
format_args!(
"Destination node {} is not a master",
String::from_utf8_lossy(&dest)
),
));
}
}
server.asm_begin_migrate(args.get(3), session.node_id(), ranges, session.row())?;
out.simple(b"RDBCHANNELSYNCSLOTS");
Ok(())
}
fn rdbchannel(server: &Server, session: &mut Session, args: Args<'_>, out: &mut Out) -> Result<()> {
let id = args.get(3);
if id.len() != ID_LEN {
return Err(Error::new(Code::Invalid, "Invalid task id"));
}
let ranges = server.asm_take_rdb_channel(id, session.row().id)?;
out.simple(b"SLOTSSNAPSHOT");
out.raw(&server.asm_snapshot(&ranges));
Ok(())
}
fn conf(server: &Server, session: &mut Session, args: Args<'_>, out: &mut Out) -> Result<()> {
let mut at = 3;
while at < args.len() {
if at + 1 >= args.len() {
super::write_error(out, &wrong_sub_arity(args.get(1)));
return Ok(());
}
let name = args.get(at);
let value = args.get(at + 1);
if args::is(name, b"node-id") {
if value.len() != ID_LEN {
let len = value.len();
super::write_error(
out,
&Error::fmt(Code::Invalid, format_args!("Invalid node id length {len}")),
);
return Ok(());
}
if server.cluster.map.lock().find(value).is_none() {
super::write_error(
out,
&Error::fmt(
Code::Invalid,
format_args!(
"Node {} not found in cluster",
String::from_utf8_lossy(value)
),
),
);
return Ok(());
}
session.set_node_id(value);
} else if args::is(name, b"slot-info") {
if !slot_info(value) {
super::write_error(
out,
&Error::fmt(
Code::Invalid,
format_args!("Invalid slot info: {}", String::from_utf8_lossy(value)),
),
);
return Ok(());
}
} else if args::is(name, b"asm-task") {
if server.cluster.map.lock().nodes[0].is_master() {
super::write_error(
out,
&Error::new(
Code::Invalid,
"CLUSTER SYNCSLOTS CONF ASM-TASK only allowed on replica",
),
);
return Ok(());
}
super::write_error(
out,
&Error::fmt(
Code::Invalid,
format_args!(
"Failed to handle master task: {}",
String::from_utf8_lossy(value)
),
),
);
} else if !args::is(name, b"capa") {
super::write_error(
out,
&Error::fmt(
Code::Invalid,
format_args!("Unknown option {}", String::from_utf8_lossy(name)),
),
);
}
at += 2;
}
out.ok();
Ok(())
}
fn slot_info(value: &[u8]) -> bool {
let mut parts = value.split(|b| *b == b':');
let Some(slot) = parts.next().and_then(yo_common::num::parse_i64) else {
return false;
};
let Some(keys) = parts.next().and_then(yo_common::num::parse_i64) else {
return false;
};
let Some(expires) = parts.next().and_then(yo_common::num::parse_i64) else {
return false;
};
parts.next().is_none() && (0..SLOTS as i64).contains(&slot) && keys >= 0 && expires >= 0
}
fn setslot(server: &Server, session_db: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
if !server.cluster.map.lock().nodes[0].is_master() {
return Err(Error::new(
Code::Invalid,
"Please use SETSLOT only with masters.",
));
}
let slot = slot_arg(&args, 2)?;
let action = args.get(3);
let wrong = || {
Error::new(
Code::Invalid,
"Invalid CLUSTER SETSLOT action or number of arguments. Try CLUSTER HELP",
)
};
let mut announce = false;
let mut follow: Option<u16> = None;
{
let mut map = server.cluster.map.lock();
let at = usize::from(slot);
if args::is(action, b"migrating") && args.len() == 5 {
if !map.mine(slot) {
return Err(Error::fmt(
Code::Invalid,
format_args!("I'm not the owner of hash slot {slot}"),
));
}
let Some(to) = map.find(args.get(4)) else {
return Err(dont_know(args.get(4)));
};
map.migrating[at] = Some(to);
} else if args::is(action, b"importing") && args.len() == 5 {
if map.mine(slot) {
return Err(Error::fmt(
Code::Invalid,
format_args!("I'm already the owner of hash slot {slot}"),
));
}
let Some(from) = map.find(args.get(4)) else {
return Err(dont_know(args.get(4)));
};
map.importing[at] = Some(from);
} else if args::is(action, b"stable") && args.len() == 4 {
map.migrating[at] = None;
map.importing[at] = None;
} else if args::is(action, b"node") && args.len() == 5 {
let Some(to) = map.find(args.get(4)) else {
return Err(unknown_node(args.get(4)));
};
if !map.nodes[usize::from(to)].is_master() {
return Err(Error::new(Code::Invalid, "Target node is not a master"));
}
let was_mine = map.owner[at] == Some(0);
let held = keys_in_slot(server, session_db, slot);
if was_mine && to != 0 && held != 0 {
return Err(Error::fmt(
Code::Invalid,
format_args!(
"Can't assign hashslot {slot} to a different node while I still hold keys for this hash slot."
),
));
}
if held == 0 {
map.migrating[at] = None;
}
map.owner[at] = Some(to);
if was_mine && to != 0 && map.runs(0).is_empty() {
follow = Some(to);
}
if to == 0 && map.importing[at].is_some() {
bump_without_consensus(server, &mut map);
map.importing[at] = None;
announce = true;
}
} else {
return Err(wrong());
}
}
server.recount_coverage();
save(server)?;
if let Some(to) = follow
&& let Some(shared) = server.myself()
{
shared.cluster_replicate(to);
}
if announce {
server.cluster_broadcast_pong();
}
out.ok();
Ok(())
}
fn keys_in_slot(server: &Server, at: usize, slot: u16) -> usize {
let mut found = 0;
server.dbs[at].keys(|key| {
if key_slot(key) == slot {
found += 1;
}
});
found
}
fn bump_without_consensus(server: &Server, map: &mut Map) -> bool {
let highest = map
.nodes
.iter()
.map(|n| n.epoch)
.max()
.unwrap_or(0)
.max(server.cluster.epoch.load(Relaxed));
let mine = map.nodes[0].epoch;
if mine != 0 && mine == highest {
return false;
}
map.nodes[0].epoch = server.cluster.epoch.fetch_add(1, Relaxed) + 1;
true
}
fn flushslots(server: &Server, out: &mut Out) -> Result<()> {
if server.dbs.iter().any(|db| !db.is_empty()) {
return Err(Error::new(
Code::Invalid,
"DB must be empty to perform CLUSTER FLUSHSLOTS.",
));
}
{
let mut map = server.cluster.map.lock();
for slot in 0..SLOTS {
if map.owner[slot] == Some(0) {
map.owner[slot] = None;
}
map.migrating[slot] = None;
map.importing[slot] = None;
}
}
server.recount_coverage();
save(server)?;
out.ok();
Ok(())
}
fn bumpepoch(server: &Server, out: &mut Out) -> Result<()> {
let (moved, epoch) = {
let mut map = server.cluster.map.lock();
let moved = bump_without_consensus(server, &mut map);
(moved, map.nodes[0].epoch)
};
if moved {
save(server)?;
server.cluster_broadcast_pong();
}
let word = if moved { "BUMPED" } else { "STILL" };
let text = yo_alloc::allow(|| format!("{word} {epoch}"));
out.simple(text.as_bytes());
Ok(())
}
fn set_config_epoch(server: &Server, args: Args<'_>, out: &mut Out) -> Result<()> {
let epoch = args.int(2)?;
if epoch < 0 {
return Err(Error::fmt(
Code::Invalid,
format_args!("Invalid config epoch specified: {epoch}"),
));
}
{
let mut map = server.cluster.map.lock();
if map.nodes[0].epoch != 0 {
return Err(Error::new(
Code::Invalid,
"Node config epoch is already non-zero",
));
}
map.nodes[0].epoch = epoch as u64;
}
let epoch = epoch as u64;
server.cluster.epoch.fetch_max(epoch, Relaxed);
save(server)?;
out.ok();
Ok(())
}
fn reset(server: &Server, args: Args<'_>, out: &mut Out) -> Result<()> {
let hard = match args.opt(2) {
None => false,
Some(word) if args::is(word, b"hard") => true,
Some(word) if args::is(word, b"soft") => false,
Some(_) => return Err(args::syntax()),
};
if server.dbs.iter().any(|db| !db.is_empty()) {
return Err(Error::new(
Code::Invalid,
"CLUSTER RESET can't be called with master nodes containing keys",
));
}
{
let mut map = server.cluster.map.lock();
for slot in 0..SLOTS {
map.owner[slot] = None;
map.migrating[slot] = None;
map.importing[slot] = None;
}
map.nodes.truncate(1);
map.nodes[0].epoch = 0;
map.nodes[0].flags = FLAG_MYSELF | FLAG_MASTER;
map.nodes[0].master = None;
if hard {
yo_alloc::allow(|| {
map.nodes[0].id = String::from_utf8_lossy(&new_id()).into_owned();
map.nodes[0].shard = String::from_utf8_lossy(&new_id()).into_owned();
});
}
}
if hard {
server.cluster.epoch.store(0, Relaxed);
}
server.recount_coverage();
save(server)?;
out.ok();
Ok(())
}
fn count_keys(server: &Server, at: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
let n = args.int(2)?;
let slot = u16::try_from(n)
.ok()
.filter(|s| usize::from(*s) < SLOTS)
.ok_or_else(|| Error::new(Code::Invalid, "Invalid slot"))?;
let mut found = 0i64;
server.dbs[at].keys(|key| {
if key_slot(key) == slot {
found += 1;
}
});
out.int(found);
Ok(())
}
fn get_keys(server: &Server, at: usize, args: Args<'_>, out: &mut Out) -> Result<()> {
let slot = args.int(2)?;
let count = args.int(3)?;
let bad = || Error::new(Code::Invalid, "Invalid slot or number of keys");
if !(0..SLOTS as i64).contains(&slot) || count < 0 {
return Err(bad());
}
let slot = slot as u16;
let want = count as usize;
let start = out.len();
let mut n = 0;
server.dbs[at].keys(|key| {
if n < want && key_slot(key) == slot {
out.bulk(key);
n += 1;
}
});
out.close_array(start, n);
Ok(())
}
fn save(server: &Server) -> Result<()> {
let path = {
let file = server.cluster.file.lock();
if file.is_empty() {
return Ok(());
}
yo_alloc::allow(|| file.clone())
};
let epoch = server.cluster.epoch.load(Relaxed);
let text = yo_alloc::allow(|| {
let mut s = lines(server, true);
let _ = writeln!(s, "vars currentEpoch {epoch} lastVoteEpoch 0");
s
});
yo_alloc::allow(|| {
let temp = format!("{path}.tmp");
let wrote =
std::fs::write(&temp, text.as_bytes()).and_then(|()| std::fs::rename(&temp, &path));
match wrote {
Ok(()) => Ok(()),
Err(e) => Err(Error::fmt(
Code::Invalid,
format_args!("cluster config file could not be written: {e}"),
)),
}
})
}
impl Server {
fn reload_cluster(&mut self) -> Result<()> {
let path = self.cluster_file();
if path.is_empty() {
return Ok(());
}
let text = match yo_alloc::allow(|| std::fs::read_to_string(&path)) {
Ok(text) => text,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
Err(e) => {
return Err(Error::fmt(Code::Invalid, format_args!("{path}: {e}")));
}
};
yo_alloc::allow(|| self.absorb_cluster(&text))
}
fn absorb_cluster(&mut self, text: &str) -> Result<()> {
let bad = |what: &str| Error::fmt(Code::Invalid, format_args!("{what} in cluster config"));
let now = self.now_ms();
let mut nodes: Vec<Node> = Vec::new();
let mut owned: Vec<(String, u16, u16)> = Vec::new();
let mut follows: Vec<(String, String)> = Vec::new();
for line in text.lines() {
let line = line.trim();
if line.is_empty() {
continue;
}
let mut words = line.split(' ');
let first = words.next().unwrap_or_default();
if first == "vars" {
let rest: Vec<&str> = words.collect();
for pair in rest.chunks(2) {
if pair.len() == 2 && pair[0] == "currentEpoch" {
let epoch = pair[1].parse::<u64>().map_err(|_| bad("bad epoch"))?;
self.cluster.epoch.store(epoch, Relaxed);
}
}
continue;
}
if first.len() != ID_LEN {
return Err(bad("bad node id"));
}
let address = words.next().ok_or_else(|| bad("missing address"))?;
let named = words.next().ok_or_else(|| bad("missing flags"))?;
let master = words.next().ok_or_else(|| bad("missing master"))?;
let ping = words.next().ok_or_else(|| bad("missing ping time"))?;
let pong = words.next().ok_or_else(|| bad("missing pong time"))?;
let epoch = words
.next()
.and_then(|w| w.parse::<u64>().ok())
.ok_or_else(|| bad("bad config epoch"))?;
let _link = words.next();
let (host, port, bus, shard) =
split_address(address).ok_or_else(|| bad("bad address"))?;
for word in words {
if word.starts_with('[') {
continue;
}
let (from, to) = match word.split_once('-') {
Some((a, b)) => (
a.parse::<u16>().map_err(|_| bad("bad slot"))?,
b.parse::<u16>().map_err(|_| bad("bad slot"))?,
),
None => {
let one = word.parse::<u16>().map_err(|_| bad("bad slot"))?;
(one, one)
}
};
if usize::from(from) >= SLOTS || usize::from(to) >= SLOTS || from > to {
return Err(bad("slot out of range"));
}
owned.push((first.to_owned(), from, to));
}
let mut flags = 0u16;
for name in named.split(',') {
flags |= match name {
"myself" => FLAG_MYSELF,
"master" => FLAG_MASTER,
"slave" => FLAG_SLAVE,
"fail?" => FLAG_PFAIL,
"fail" => FLAG_FAIL,
"handshake" => FLAG_HANDSHAKE,
"noaddr" => FLAG_NOADDR,
"nofailover" => FLAG_NOFAILOVER,
_ => 0,
};
}
if master != "-" {
if master.len() != ID_LEN {
return Err(bad("bad master id"));
}
follows.push((first.to_owned(), master.to_owned()));
}
let node = Node {
id: first.to_owned(),
host,
port,
bus,
shard,
epoch,
flags,
master: None,
ping_sent: stamp(ping, now),
pong_recv: stamp(pong, now),
data_recv: now,
fail_time: 0,
offset: 0,
linked: false,
reports: Vec::new(),
};
if named.split(',').any(|f| f == "myself") {
nodes.insert(0, node);
} else {
nodes.push(node);
}
}
if nodes.is_empty() {
return Ok(());
}
let mut map = Map::new(nodes.remove(0));
map.nodes.append(&mut nodes);
for (id, from, to) in owned {
let Some(at) = map.find(id.as_bytes()) else {
return Err(bad("slots for a node nobody knows"));
};
for slot in from..=to {
map.owner[usize::from(slot)] = Some(at);
}
}
for (who, whose) in follows {
let (Some(who), Some(whose)) = (map.find(who.as_bytes()), map.find(whose.as_bytes()))
else {
return Err(bad("master id nobody knows"));
};
map.nodes[usize::from(who)].master = Some(whose);
map.nodes[usize::from(who)].epoch = 0;
}
*self.cluster.map.lock() = map;
Ok(())
}
}
fn stamp(field: &str, now: u64) -> u64 {
match field.parse::<u64>() {
Ok(0) | Err(_) => 0,
Ok(_) => now,
}
}
fn split_address(field: &str) -> Option<(String, u16, u16, String)> {
let (address, aux) = match field.split_once(',') {
Some((address, aux)) => (address, aux),
None => (field, ""),
};
let (host, ports) = address.rsplit_once(':')?;
let (client, bus) = match ports.split_once('@') {
Some((client, bus)) => (client, bus.parse::<u16>().ok()?),
None => (ports, 0),
};
let port = client.parse::<u16>().ok()?;
let bus = if bus == 0 { port + BUS_OFFSET } else { bus };
let shard = aux
.split(',')
.find_map(|pair| pair.strip_prefix("shard-id="))
.map_or_else(
|| String::from_utf8_lossy(&new_id()).into_owned(),
str::to_owned,
);
Some((host.to_owned(), port, bus, shard))
}
#[cfg(test)]
impl Server {
pub(super) fn cluster_own_everything(&self) {
{
let mut map = self.cluster.map.lock();
for slot in 0..SLOTS {
map.owner[slot] = Some(0);
}
}
self.recount_coverage();
self.cluster.was_down.store(false, Relaxed);
self.cluster.booted_at.store(0, Relaxed);
}
pub(super) fn cluster_pretend_node(&self, id: &str, host: &str, port: u16) -> u16 {
let mut map = self.cluster.map.lock();
let mut node = Node::new(
id.to_owned(),
host.to_owned(),
port,
port + BUS_OFFSET,
FLAG_MASTER,
0,
);
node.shard = id.to_owned();
node.linked = true;
map.nodes.push(node);
(map.nodes.len() - 1) as u16
}
pub(super) fn cluster_hand_over(&self, slot: u16, node: u16) {
let mut map = self.cluster.map.lock();
map.owner[usize::from(slot)] = Some(node);
}
pub(super) fn cluster_pretend_follower(&self, of: u16) {
let mut map = self.cluster.map.lock();
map.nodes[0].flags &= !FLAG_MASTER;
map.nodes[0].flags |= FLAG_SLAVE;
map.nodes[0].master = Some(of);
}
pub(super) fn cluster_moving(&self, slot: u16, to: Option<u16>, from: Option<u16>) {
let mut map = self.cluster.map.lock();
map.migrating[usize::from(slot)] = to;
map.importing[usize::from(slot)] = from;
}
}
#[cfg(test)]
mod tests {
use super::{SLOTS, key_slot};
#[test]
fn a_key_lands_in_the_slot_a_real_server_puts_it_in() {
assert_eq!(key_slot(b"foo"), 12182);
assert_eq!(key_slot(b"1234"), 6025);
assert_eq!(key_slot(b""), 0);
assert_eq!(key_slot(b"{user1000}.following"), 3443);
}
#[test]
fn the_hash_tag_rules_are_the_reference_rules() {
assert_eq!(
key_slot(b"{user1000}.following"),
key_slot(b"{user1000}.followers")
);
assert_eq!(key_slot(b"{}foo"), key_slot(b"{}foo"));
assert_ne!(key_slot(b"{}foo"), key_slot(b"foo"));
assert_ne!(key_slot(b"{foo"), key_slot(b"foo"));
assert_eq!(key_slot(b"{a}{b}"), key_slot(b"a"));
assert_eq!(key_slot(b"foo{{bar}}zap"), key_slot(b"{bar"));
}
#[test]
fn every_slot_is_in_range() {
let mut seen = vec![false; SLOTS];
for i in 0..200_000u32 {
let key = i.to_string();
let slot = key_slot(key.as_bytes());
assert!(usize::from(slot) < SLOTS);
seen[usize::from(slot)] = true;
}
assert!(seen.iter().all(|s| *s), "200k keys reach all 16384 slots");
}
#[test]
fn a_config_file_puts_every_run_on_the_node_that_owns_it() {
let mut server = super::Server::new();
server.enable_cluster("", 7355);
let text = "\
3b80b05445f38bc7214f083696a2bbf90e3f30e3 127.0.0.1:7356@17356 master - 0 0 0 connected 10923-16383
30d0651b0ec5e178e082634c44fb9adcc6e4021b 127.0.0.1:7355@17355 myself,master - 0 0 1 connected 5461-10922
19a9e69b8b66016ac43c55ccdeed0283e0148e17 127.0.0.1:7354@17354 master - 0 0 2 connected 0-5460
vars currentEpoch 2 lastVoteEpoch 0
";
server.absorb_cluster(text).expect("the file parses");
let map = server.cluster.map.lock();
let at = |id: &str| map.find(id.as_bytes()).expect("the node is in the table");
assert_eq!(at("30d0651b0ec5e178e082634c44fb9adcc6e4021b"), 0, "myself");
for (id, from, to) in [
("19a9e69b8b66016ac43c55ccdeed0283e0148e17", 0, 5460),
("30d0651b0ec5e178e082634c44fb9adcc6e4021b", 5461, 10922),
("3b80b05445f38bc7214f083696a2bbf90e3f30e3", 10923, 16383),
] {
let owner = Some(at(id));
for slot in from..=to {
assert_eq!(map.owner[slot], owner, "slot {slot} belongs to {id}");
}
}
}
#[test]
fn a_config_file_is_not_read_back_as_live_state() {
let mut server = super::Server::new();
server.enable_cluster("", 7357);
let text = "\
9fcbb7624dedbb2fd0020dd2fbf86a5eb8cec31b 127.0.0.1:7355@17355 master - 0 1789005531306 3 connected 5461-10922
ac6dd51a69741dc5130637c594866c9b7e0cfc4e 127.0.0.1:7357@17357 myself,slave 9fcbb7624dedbb2fd0020dd2fbf86a5eb8cec31b 1789005400000 1789005532315 3 connected
";
server.absorb_cluster(text).expect("the file parses");
let now = server.now_ms();
let map = server.cluster.map.lock();
for node in &map.nodes {
assert!(!node.linked, "nothing is linked before the bus dials out");
}
assert_eq!(map.nodes[0].ping_sent, now, "myself had a ping in flight");
assert_eq!(map.nodes[1].ping_sent, 0, "the master did not");
assert_eq!(map.nodes[0].pong_recv, now);
assert_eq!(map.nodes[0].epoch, 0, "the replica's epoch is dropped");
assert_eq!(map.nodes[1].epoch, 3, "the master's is kept");
}
}