surrealdb_core/dbs/
node.rs

1use std::fmt::{self, Display};
2use std::ops::{Add, Sub};
3use std::time::Duration;
4
5use revision::revisioned;
6use serde::{Deserialize, Serialize};
7use surrealdb_types::{SqlFormat, ToSql, write_sql};
8use uuid::Uuid;
9
10use crate::expr::statements::info::InfoStructure;
11use crate::kvs::impl_kv_value_revisioned;
12use crate::val::{Object, Value};
13
14/// A node in the cluster
15#[revisioned(revision = 2)]
16#[derive(Clone, Debug, Default, Eq, PartialEq, PartialOrd, Serialize, Deserialize, Hash)]
17pub struct Node {
18	/// The id of the node
19	pub id: Uuid,
20	/// The heartbeat of the node
21	pub heartbeat: Timestamp,
22	/// Whether the node is garbage collected
23	pub gc: bool,
24	/// Public HTTP endpoint at which other cluster members can reach this node
25	/// for cross-node messaging (live-query relay, etc.). `None` when not
26	/// configured — single-node and shared-backend deployments leave this empty.
27	#[revision(start = 2)]
28	pub http_endpoint: Option<String>,
29}
30
31impl_kv_value_revisioned!(Node);
32
33impl Node {
34	/// Create a new Node entry
35	pub fn new(id: Uuid, hb: Timestamp, gc: bool) -> Self {
36		Self {
37			id,
38			heartbeat: hb,
39			gc,
40			http_endpoint: None,
41		}
42	}
43
44	/// Create a new Node entry with a public HTTP endpoint for cross-node
45	/// messaging.
46	pub fn new_with_endpoint(
47		id: Uuid,
48		hb: Timestamp,
49		gc: bool,
50		http_endpoint: Option<String>,
51	) -> Self {
52		Self {
53			id,
54			heartbeat: hb,
55			gc,
56			http_endpoint,
57		}
58	}
59	/// Mark this node as archived
60	pub fn archive(&self) -> Self {
61		Node {
62			gc: true,
63			..self.to_owned()
64		}
65	}
66	/// Check if this node is active
67	pub fn id(&self) -> Uuid {
68		self.id
69	}
70	/// Check if this node is active
71	pub fn is_active(&self) -> bool {
72		!self.gc
73	}
74	/// Check if this node is archived
75	pub fn is_archived(&self) -> bool {
76		self.gc
77	}
78	// Return the node id if archived
79	pub fn archived(&self) -> Option<Uuid> {
80		self.is_archived().then_some(self.id)
81	}
82}
83
84impl ToSql for Node {
85	fn fmt_sql(&self, f: &mut String, sql_fmt: SqlFormat) {
86		write_sql!(f, sql_fmt, "NODE {} SEEN {}", self.id, self.heartbeat);
87		if self.gc {
88			write_sql!(f, sql_fmt, " ARCHIVED");
89		} else {
90			write_sql!(f, sql_fmt, " ACTIVE");
91		}
92	}
93}
94
95impl InfoStructure for Node {
96	fn structure(self) -> Value {
97		let object = map! {
98			"id" => Value::Uuid(self.id.into()),
99			"seen" => self.heartbeat.structure(),
100			"active" => Value::Bool(!self.gc),
101		};
102		Value::Object(Object::from(object))
103	}
104}
105
106// This struct is meant to represent a timestamp that can be used to partially
107// order events in a cluster. It should be derived from a timestamp oracle, such
108// as the one available in TiKV via the client `TimestampExt` implementation.
109#[revisioned(revision = 1)]
110#[derive(Clone, Copy, Default, Debug, Eq, PartialEq, PartialOrd, Deserialize, Serialize, Hash)]
111pub struct Timestamp {
112	pub value: u64,
113}
114
115impl From<u64> for Timestamp {
116	fn from(value: u64) -> Self {
117		Timestamp {
118			value,
119		}
120	}
121}
122
123impl Add<Duration> for Timestamp {
124	type Output = Timestamp;
125	fn add(self, rhs: Duration) -> Self::Output {
126		Timestamp {
127			value: self.value.wrapping_add(rhs.as_millis() as u64),
128		}
129	}
130}
131
132impl Sub<Duration> for Timestamp {
133	type Output = Timestamp;
134	fn sub(self, rhs: Duration) -> Self::Output {
135		Timestamp {
136			value: self.value.wrapping_sub(rhs.as_millis() as u64),
137		}
138	}
139}
140
141impl Display for Timestamp {
142	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
143		write!(f, "{}", self.value)
144	}
145}
146
147impl ToSql for Timestamp {
148	fn fmt_sql(&self, f: &mut String, fmt: SqlFormat) {
149		write_sql!(f, fmt, "{}", self.value)
150	}
151}
152
153impl InfoStructure for Timestamp {
154	fn structure(self) -> Value {
155		self.value.into()
156	}
157}
158
159#[cfg(test)]
160mod test {
161	use std::time::Duration;
162
163	use chrono::TimeZone;
164	use chrono::prelude::Utc;
165
166	use crate::dbs::node::Timestamp;
167
168	#[test]
169	fn timestamps_can_be_added_duration() {
170		let t = Utc.with_ymd_and_hms(2000, 1, 1, 12, 30, 0).unwrap();
171		let ts = Timestamp {
172			value: t.timestamp_millis() as u64,
173		};
174
175		let hour = Duration::from_secs(60 * 60);
176		let ts = ts + hour;
177		let ts = ts + hour;
178		let ts = ts + hour;
179
180		let end_time = Utc.timestamp_millis_opt(ts.value as i64).unwrap();
181		let expected_end_time = Utc.with_ymd_and_hms(2000, 1, 1, 15, 30, 0).unwrap();
182		assert_eq!(end_time, expected_end_time);
183	}
184
185	#[test]
186	fn timestamps_can_be_subtracted_duration() {
187		let t = Utc.with_ymd_and_hms(2000, 1, 1, 12, 30, 0).unwrap();
188		let ts = Timestamp {
189			value: t.timestamp_millis() as u64,
190		};
191
192		let hour = Duration::from_secs(60 * 60);
193		let ts = ts - hour;
194		let ts = ts - hour;
195		let ts = ts - hour;
196
197		let end_time = Utc.timestamp_millis_opt(ts.value as i64).unwrap();
198		let expected_end_time = Utc.with_ymd_and_hms(2000, 1, 1, 9, 30, 0).unwrap();
199		assert_eq!(end_time, expected_end_time);
200	}
201}
surrealdb_core/dbs/node.rs

surrealdb_core/dbs/
node.rs
