wedb_standalone 0.1.2

Standalone server foundation: RESP protocol, AOF logic layer, and node service orchestration
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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
//! 集群会话抽象面(对标 libs/server/Cluster/IClusterSession.cs;语义对标
//! libs/cluster/Session/ClusterSession.cs 及 SlotVerification/ 分片)
//!
//! C# 为接口 + 分布式实现(ClusterSession:gossip / 迁移 / 复制回放驱动);
//! 本域按单机 provider 语义对标(见 [`super::i_cluster_provider`] 头注):
//! 角色恒为主(`ReadWriteSession` 经 `CurrentConfig.IsPrimary` 短路恒真)、
//! 全部槽位归属本节点(单键验证恒 OK,MOVED/ASK/CLUSTERDOWN 不可能)、
//! 集群命令族(MIGRATE / FAILOVER / REPLICAOF / CLUSTER 子命令)与主复制
//! 流为直通空操作。迭代式槽位验证状态机按 C# 原样保留——单机形态下其
//! 判定恒通过,但跨槽 / 重试状态迁移逻辑与分布式实现逐位一致,供上层在
//! 接线分布式 provider 时零改动复用。

use std::sync::Arc;

use crate::{
  acl::user_handle::UserHandle,
  storage::session::common::array_key_iteration_functions::cluster_slot, types::RespCommand,
};

/// 单机形态的当前纪元(C# clusterProvider.GarnetCurrentEpoch 的单机承接;
/// 分布式 epoch 由 ClusterConfig 版本化,单机恒为 1)
pub const SINGLE_NODE_CURRENT_EPOCH: i64 = 1;

/// 槽位验证结果状态(对标 libs/cluster/Session/SlotVerification/ClusterSlotVerificationResult.cs
/// 引用的 SlotVerifiedState)
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SlotVerifiedState {
  /// 验证通过
  Ok,
  /// 多键请求跨槽
  CrossSlot,
  /// 槽位迁移中需重试
  TryAgain,
  /// 槽位归属他节点
  Moved,
}

/// 迭代式验证的缓存结果(C# cachedVerificationResult:状态 + 槽位)
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct CachedSlotVerification {
  state: SlotVerifiedState,
  slot: u16,
}

/// 缓存槽位验证错误应答文案(C# CmdStrings.RESP_ERR_CROSSSLOT / RESP_ERR_TRYAGAIN)
const RESP_ERR_CROSSSLOT: &str = "CROSSSLOT Keys in request don't hash to the same slot";
const RESP_ERR_TRYAGAIN: &str = "TRYAGAIN Multiple keys request during rehashing of slot";

/// 集群 RESP 会话(单机 provider 语义)
pub struct IClusterSession {
  /// 最近一次 GOSSIP 呈现的远端节点 id(单机无 gossip:恒 None)
  remote_node_id: Option<String>,
  /// 读 / 写会话标志(C# readWriteSession 字段)
  read_write_session: bool,
  /// 是否处于活动复制流(首个 APPENDLOG 握手后置位;单机恒 false)
  is_replicating: bool,
  /// 本地当前纪元(Acquire/Release 配对维护)
  local_current_epoch: i64,
  /// 会话当前鉴权用户(权限检查用)
  user_handle: Option<Arc<UserHandle>>,
  /// 迭代式验证缓存(initialized = C# `initialized` 字段)
  cached_verification: Option<CachedSlotVerification>,
}

impl Default for IClusterSession {
  fn default() -> Self {
    Self::new()
  }
}

impl IClusterSession {
  /// 构造单机集群会话(C# 构造子装配子集:无 provider / txnManager /
  /// 网络发送器依赖,验证缓存未初始化)
  pub fn new() -> Self {
    Self {
      remote_node_id: None,
      read_write_session: false,
      is_replicating: false,
      local_current_epoch: 0,
      user_handle: None,
      cached_verification: None,
    }
  }

  /// libs/server/Cluster/IClusterSession.cs:RemoteNodeId
  ///
  /// 最近一次 GOSSIP 呈现的远端节点 id(单机恒 None)
  pub fn remote_node_id(&self) -> Option<&str> {
    self.remote_node_id.as_deref()
  }

  /// libs/server/Cluster/IClusterSession.cs:ReadWriteSession
  ///
  /// C# `CurrentConfig.IsPrimary || readWriteSession`——单机恒为主,
  /// 本属性恒真(标志位本体见 [`Self::read_write_session_flag`])
  pub fn read_write_session(&self) -> bool {
    true
  }

  /// 读 / 写会话标志位本体(SetReadOnly/SetReadWrite 的直接产物)
  pub fn read_write_session_flag(&self) -> bool {
    self.read_write_session
  }

  /// libs/server/Cluster/IClusterSession.cs:IsReplicating
  ///
  /// 是否处于活动复制流(单机无复制流:恒 false)
  pub fn is_replicating(&self) -> bool {
    self.is_replicating
  }

  /// libs/server/Cluster/IClusterSession.cs:LocalCurrentEpoch
  pub fn local_current_epoch(&self) -> i64 {
    self.local_current_epoch
  }

  /// libs/server/Cluster/IClusterSession.cs:SetReadOnlySession
  pub fn set_read_only_session(&mut self) {
    self.read_write_session = false;
  }

  /// libs/server/Cluster/IClusterSession.cs:SetReadWriteSession
  pub fn set_read_write_session(&mut self) {
    self.read_write_session = true;
  }

  /// libs/server/Cluster/IClusterSession.cs:AcquireCurrentEpoch
  ///
  /// 采样当前纪元到本地(单机纪元恒 [`SINGLE_NODE_CURRENT_EPOCH`])
  pub fn acquire_current_epoch(&mut self) {
    self.local_current_epoch = SINGLE_NODE_CURRENT_EPOCH;
  }

  /// libs/server/Cluster/IClusterSession.cs:ReleaseCurrentEpoch
  pub fn release_current_epoch(&mut self) {
    self.local_current_epoch = 0;
  }

  /// libs/server/Cluster/IClusterSession.cs:ProcessClusterCommands
  ///
  /// 集群命令族分派(CLUSTER 子命令 / MIGRATE / FAILOVER / REPLICAOF /
  /// SECONDARYOF)。单机无集群命令面:直通空操作并返回 false(未处理,
  /// 调用方按普通命令继续),与 C# `!EnableCluster` 时 clusterSession
  /// 不接线的语义一致。命令参数仅作入参形态承接
  pub fn process_cluster_commands(&mut self, _command: RespCommand) -> bool {
    false
  }

  /// libs/server/Cluster/IClusterSession.cs:ResetCachedSlotVerificationResult
  ///
  /// 迭代式验证缓存复位(调用方在每个多键命令验证前调用)
  pub fn reset_cached_slot_verification_result(&mut self) {
    self.cached_verification = None;
  }

  /// libs/server/Cluster/IClusterSession.cs:NetworkIterativeSlotVerify
  ///
  /// 迭代式单键验证(跨调用缓存结果;调用方须配对 Reset):
  /// 首次验证初始化缓存;缓存非 OK 提前短路(捕获首错);槽位变化 →
  /// CROSSSLOT;状态变化 → TRYAGAIN。单机全槽位自有 → 单键验证恒 OK,
  /// 故正常路径恒真
  pub fn network_iterative_slot_verify(
    &mut self,
    key: &[u8],
    read_only: bool,
    session_asking: u8,
    wait_for_stable_slot: bool,
  ) -> bool {
    let slot = cluster_slot(key);
    let verify = CachedSlotVerification {
      state: self.verify_single_key(slot, read_only, session_asking != 0, wait_for_stable_slot),
      slot,
    };

    match self.cached_verification {
      None => {
        self.cached_verification = Some(verify);
        verify.state == SlotVerifiedState::Ok
      }
      Some(cached) if cached.state != SlotVerifiedState::Ok => false,
      Some(cached) => {
        if verify.slot != cached.slot {
          self.cached_verification = Some(CachedSlotVerification {
            state: SlotVerifiedState::CrossSlot,
            slot: cached.slot,
          });
          return false;
        }
        if verify.state != cached.state {
          self.cached_verification = Some(CachedSlotVerification {
            state: SlotVerifiedState::TryAgain,
            slot: cached.slot,
          });
          return false;
        }
        verify.state == SlotVerifiedState::Ok
      }
    }
  }

  /// 单键槽位归属验证(C# SingleKeySlotVerify 的单机承接):
  /// 全部槽位归属本节点 → 恒 OK;MOVED/ASK/CLUSTERDOWN 仅分布式形态可达
  fn verify_single_key(
    &self,
    _slot: u16,
    _read_only: bool,
    _session_asking: bool,
    _wait_for_stable_slot: bool,
  ) -> SlotVerifiedState {
    SlotVerifiedState::Ok
  }

  /// libs/server/Cluster/IClusterSession.cs:WriteCachedSlotVerificationMessage
  ///
  /// 缓存验证失败时写错误应答(OK 时不写任何字节)。单机可达状态仅
  /// CROSSSLOT / TRYAGAIN(MOVED 归属分布式)
  pub fn write_cached_slot_verification_message(&self, output: &mut Vec<u8>) {
    if let Some(cached) = self
      .cached_verification
      .filter(|c| c.state != SlotVerifiedState::Ok)
    {
      let message = match cached.state {
        SlotVerifiedState::CrossSlot => RESP_ERR_CROSSSLOT,
        SlotVerifiedState::TryAgain => RESP_ERR_TRYAGAIN,
        SlotVerifiedState::Moved => "MOVED",
        SlotVerifiedState::Ok => unreachable!("filtered above"),
      };
      output.push(b'-');
      output.extend_from_slice(message.as_bytes());
      output.extend_from_slice(b"\r\n");
    }
  }

  /// libs/server/Cluster/IClusterSession.cs:NetworkMultiKeySlotVerify
  ///
  /// 多键槽位验证(验证失败写应答并返回 true;通过返回 false 继续)。
  /// C# 跳过条件:`!EnableCluster || 事务运行中` → false;单机形态
  /// EnableCluster 恒 false → 恒跳过验证
  pub fn network_multi_key_slot_verify(
    &mut self,
    _keys: &[&[u8]],
    _is_txn: bool,
    _output: &mut Vec<u8>,
  ) -> bool {
    false
  }

  /// libs/server/Cluster/IClusterSession.cs:NetworkMultiKeySlotVerifyNoResponse
  ///
  /// 多键槽位验证的无应答变体(true = 验证失败,应答交由调用方)。
  /// 单机形态恒 false(同上跳过语义)
  pub fn network_multi_key_slot_verify_no_response(
    &mut self,
    _keys: &[&[u8]],
    _is_txn: bool,
  ) -> bool {
    false
  }

  /// libs/server/Cluster/IClusterSession.cs:SetUserHandle
  ///
  /// 更新会话当前鉴权用户(ACL 权限检查用)
  pub fn set_user_handle(&mut self, user_handle: Arc<UserHandle>) {
    self.user_handle = Some(user_handle);
  }

  /// 当前鉴权用户句柄
  pub fn user_handle(&self) -> Option<&Arc<UserHandle>> {
    self.user_handle.as_ref()
  }

  /// libs/server/Cluster/IClusterSession.cs:ProcessPrimaryStream
  ///
  /// 主复制流回放入口(微基准挂钩;分布式经 ReplicaReplayDriver 承接)。
  /// 单机无主复制流:直通空操作
  pub fn process_primary_stream(
    &mut self,
    _physical_sublog_idx: usize,
    _record: &[u8],
    _previous_address: i64,
    _current_address: i64,
    _next_address: i64,
  ) {
  }

  /// libs/server/Cluster/IClusterSession.cs:Dispose
  ///
  /// 释放会话资源(C# 另含迁移接收状态与副本回放驱动的释放;单机无
  /// 此类资源,复位状态即可)
  pub fn dispose(&mut self) {
    self.is_replicating = false;
    self.cached_verification = None;
    self.remote_node_id = None;
  }
}

#[cfg(test)]
mod tests {
  use super::*;
  use crate::acl::user::User;

  #[test]
  fn read_write_semantics_follow_primary_short_circuit() {
    let mut s = IClusterSession::new();
    // C# IsPrimary || readWriteSession:单机恒为主 → 可写恒真
    assert!(s.read_write_session());
    assert!(!s.read_write_session_flag());
    s.set_read_write_session();
    assert!(s.read_write_session_flag());
    // 置只读后标志位翻转,但单机可写语义仍为真
    s.set_read_only_session();
    assert!(!s.read_write_session_flag());
    assert!(s.read_write_session());
  }

  #[test]
  fn epoch_acquire_release_roundtrip() {
    let mut s = IClusterSession::new();
    assert_eq!(s.local_current_epoch(), 0);
    s.acquire_current_epoch();
    assert_eq!(s.local_current_epoch(), SINGLE_NODE_CURRENT_EPOCH);
    s.release_current_epoch();
    assert_eq!(s.local_current_epoch(), 0);
  }

  #[test]
  fn cluster_command_family_is_passthrough_false() {
    let mut s = IClusterSession::new();
    // 单机无集群命令面:MIGRATE / CLUSTER 子命令均未处理
    assert!(!s.process_cluster_commands(RespCommand::None));
    assert!(!s.process_cluster_commands(RespCommand::Migrate));
  }

  #[test]
  fn iterative_verify_passes_same_slot_and_flags_cross_slot() {
    let mut s = IClusterSession::new();
    s.reset_cached_slot_verification_result();
    // 同哈希标签(同槽位)的多键逐键验证恒真(单机全槽位自有)
    assert!(s.network_iterative_slot_verify(b"user:{100}", true, 0, false));
    assert!(s.network_iterative_slot_verify(b"order:{100}", true, 0, false));

    // 跨槽(不同槽位):单键验证仍 OK,但缓存按 C# 状态机进入 CROSSSLOT,
    // 本键及后续键均返回 false,缓存消息给出 CROSSSLOT 应答
    let slot = cluster_slot(b"user:{100}");
    let mut probe = 0u32;
    let other_key = loop {
      let key = format!("probe-{probe}");
      if cluster_slot(key.as_bytes()) != slot {
        break key;
      }
      probe += 1;
    };
    assert!(!s.network_iterative_slot_verify(other_key.as_bytes(), true, 0, false));
    assert!(!s.network_iterative_slot_verify(b"another", true, 0, false));
    let mut out = Vec::new();
    s.write_cached_slot_verification_message(&mut out);
    assert!(String::from_utf8_lossy(&out).contains("CROSSSLOT"));
  }

  #[test]
  fn cached_verification_message_silent_on_ok() {
    let mut s = IClusterSession::new();
    s.reset_cached_slot_verification_result();
    assert!(s.network_iterative_slot_verify(b"k", false, 0, false));
    let mut out = Vec::new();
    s.write_cached_slot_verification_message(&mut out);
    // OK 缓存:不写任何字节
    assert!(out.is_empty());
  }

  #[test]
  fn reset_clears_cached_state() {
    let mut s = IClusterSession::new();
    s.reset_cached_slot_verification_result();
    assert!(s.network_iterative_slot_verify(b"k", false, 0, false));
    s.reset_cached_slot_verification_result();
    // 复位后重新以首键初始化缓存
    assert!(s.network_iterative_slot_verify(b"fresh", false, 0, false));
  }

  #[test]
  fn multi_key_verify_skipped_without_cluster() {
    let mut s = IClusterSession::new();
    let keys = [b"k1".as_slice(), b"k2".as_slice()];
    let mut out = Vec::new();
    // !EnableCluster → 跳过验证:返回 false(继续命令)且不写应答
    assert!(!s.network_multi_key_slot_verify(&keys, false, &mut out));
    assert!(out.is_empty());
    assert!(!s.network_multi_key_slot_verify_no_response(&keys, true));
  }

  #[test]
  fn primary_stream_and_remote_id_are_single_node_noops() {
    let mut s = IClusterSession::new();
    assert_eq!(s.remote_node_id(), None);
    assert!(!s.is_replicating());
    // 主复制流直通空操作
    s.process_primary_stream(0, b"record", 0, 64, 128);
    assert!(!s.is_replicating());
  }

  #[test]
  fn user_handle_swap_and_dispose() {
    let mut s = IClusterSession::new();
    assert!(s.user_handle().is_none());
    s.set_user_handle(Arc::new(UserHandle::new(Arc::new(User::new(
      "default".to_string(),
    )))));
    assert!(s.user_handle().is_some());
    s.dispose();
    assert!(s.user_handle().is_some());
    assert!(!s.is_replicating());
    assert_eq!(s.remote_node_id(), None);
  }
}