feat(recovery): implement recovery mechanism for pending entries and enhance error handling

Modified lib/infra/actor/src/actor/host/activation.rs +1 −1 View lib/infra/actor/src/actor/host/activation.rs

@@ -73,7 +73,7 @@

7373 result = work => result,

7474 };

7575 if let Some(entry) = active_entry {

76 self.inner.transport.inflight.lock().await.remove(&entry);

76 self.release(&entry).await;

7777 }

7878 let _ = tokio::time::timeout(lease_period, actor.deactivate(&ctx)).await;

7979 let shutting_down = *shutdown.borrow();

Modified lib/infra/actor/src/actor/host/builder.rs +7 −1 View lib/infra/actor/src/actor/host/builder.rs

@@ -11,7 +11,7 @@

1111use super::super::id::ActorId;

1212use super::super::redis_conn;

1313use super::super::transport::RedisTransport;

14use super::core::{ActorFactory, ActorHost, Inner};

14use super::core::{ActorFactory, ActorHost, Inner, Recovery};

1515use super::node;

1616use super::options::ActorHostOptions;

1717

@@ -92,6 +92,12 @@

9292 workers: Mutex::new(Vec::new()),

9393 events,

9494 event_receiver: Mutex::new(Some(event_receiver)),

95 recovery: Mutex::new(Recovery {

96 cursor: "0-0".to_string(),

97 needed: true,

98 generation: 0,

99 last_pass: None,

100 }),

95101 });

96102 let host = ActorHost { inner };

97103 let reader = host.clone();

Modified lib/infra/actor/src/actor/host/core.rs +41 −7 View lib/infra/actor/src/actor/host/core.rs

@@ -1,6 +1,6 @@

11use std::collections::HashMap;

22use std::sync::Arc;

3use std::time::Duration;

3use std::time::{Duration, Instant};

44

55use serde_json::Value;

66use tokio::sync::{Mutex, mpsc, oneshot, watch};

@@ -29,6 +29,14 @@

2929 pub(super) workers: Mutex<Vec<JoinHandle<()>>>,

3030 pub(super) events: mpsc::Sender<ActorEvent>,

3131 pub(super) event_receiver: Mutex<Option<mpsc::Receiver<ActorEvent>>>,

32 pub(super) recovery: Mutex<Recovery>,

33}

34

35pub(super) struct Recovery {

36 pub(super) cursor: String,

37 pub(super) needed: bool,

38 pub(super) generation: u64,

39 pub(super) last_pass: Option<Instant>,

3240}

3341

3442pub struct ActorEvent {

@@ -107,12 +115,18 @@

107115 }

108116

109117 pub async fn retry_event(&self, event: &ActorEvent) {

110 self.inner

111 .transport

112 .inflight

113 .lock()

114 .await

115 .remove(&event.entry);

118 self.release(&event.entry).await;

119 }

120

121 pub(super) async fn release(&self, entry: &str) {

122 self.inner.transport.inflight.lock().await.remove(entry);

123 let mut recovery = self.inner.recovery.lock().await;

124 recovery.needed = true;

125 recovery.generation = recovery.generation.wrapping_add(1);

126 let start = rewind_point(entry);

127 if id_before(&start, &recovery.cursor) {

128 recovery.cursor = start;

129 }

116130 }

117131

118132 pub async fn socket_heartbeat(&self, user_id: i64) -> Result<(), ActorError> {

@@ -191,3 +205,23 @@

191205 self.dispatch(id, payload, None, None).await

192206 }

193207}

208

209fn rewind_point(entry: &str) -> String {

210 match parse_entry_id(entry) {

211 Some((millis, 0)) if millis > 0 => format!("{}-{}", millis - 1, u64::MAX),

212 Some((millis, sequence)) => format!("{millis}-{}", sequence.saturating_sub(1)),

213 None => "0-0".to_string(),

214 }

215}

216

217fn id_before(left: &str, right: &str) -> bool {

218 match (parse_entry_id(left), parse_entry_id(right)) {

219 (Some(left), Some(right)) => left < right,

220 _ => true,

221 }

222}

223

224fn parse_entry_id(id: &str) -> Option<(u64, u64)> {

225 let (millis, sequence) = id.split_once('-')?;

226 Some((millis.parse().ok()?, sequence.parse().ok()?))

227}

Modified lib/infra/actor/src/actor/host/reader.rs +70 −10 View lib/infra/actor/src/actor/host/reader.rs

@@ -1,24 +1,50 @@

1use std::time::Duration;

1use std::time::{Duration, Instant};

22

33use super::super::error::ActorError;

44use super::super::transport::WireMessage;

55use super::core::{ActorEvent, ActorHost, LocalRequest, ReplyTarget};

66

77const ERROR_BACKOFF: Duration = Duration::from_millis(250);

8const MAX_ERROR_BACKOFF: Duration = Duration::from_secs(8);

9const RECOVERY_INTERVAL: Duration = Duration::from_secs(1);

810

911impl ActorHost {

1012 pub(super) async fn run_reader(self) {

1113 let mut shutdown = self.inner.shutdown.subscribe();

1214 let reader = async {

15 let mut failures = 0u32;

1316 loop {

14 self.recover_pending().await;

17 if self.recovery_due().await {

18 if let Err(error) = self.recover_pending().await {

19 failures = failures.saturating_add(1);

20 tracing::warn!(

21 node = %self.node_id(),

22 failures,

23 %error,

24 "actor reader failed to recover pending entries"

25 );

26 tokio::time::sleep(backoff(failures)).await;

27 continue;

28 }

29 failures = 0;

30 }

1531 match self.inner.transport.receive().await {

1632 Ok(entries) => {

33 failures = 0;

1734 for (entry, wire) in entries {

1835 self.handle_entry(entry, wire).await;

1936 }

2037 }

21 Err(_) => tokio::time::sleep(ERROR_BACKOFF).await,

38 Err(error) => {

39 failures = failures.saturating_add(1);

40 tracing::warn!(

41 node = %self.node_id(),

42 failures,

43 %error,

44 "actor reader failed to receive inbox entries"

45 );

46 tokio::time::sleep(backoff(failures)).await;

47 }

2248 }

2349 }

2450 };

@@ -32,23 +58,52 @@

3258 }

3359 }

3460

35 async fn recover_pending(&self) {

36 let mut after = "0-0".to_string();

61 async fn recovery_due(&self) -> bool {

62 let recovery = self.inner.recovery.lock().await;

63 recovery.needed

64 && recovery

65 .last_pass

66 .is_none_or(|at| at.elapsed() >= RECOVERY_INTERVAL)

67 }

68

69 async fn recover_pending(&self) -> Result<(), ActorError> {

70 let (mut after, generation) = {

71 let recovery = self.inner.recovery.lock().await;

72 (recovery.cursor.clone(), recovery.generation)

73 };

3774

3875 loop {

3976 let entries = match self.inner.transport.receive_pending(&after).await {

4077 Ok(entries) => entries,

41 Err(_) => return,

78 Err(error) => {

79 self.inner.recovery.lock().await.last_pass = Some(Instant::now());

80 return Err(error);

81 }

4282 };

4383 let Some(last) = entries.last().map(|(entry, _)| entry.clone()) else {

44 return;

84 break;

4585 };

4686 after = last;

4787

4888 for (entry, wire) in entries {

4989 self.handle_entry(entry, wire).await;

5090 }

91

92 let mut recovery = self.inner.recovery.lock().await;

93 recovery.last_pass = Some(Instant::now());

94 if recovery.generation != generation {

95 return Ok(());

96 }

97 recovery.cursor = after.clone();

98 }

99

100 let mut recovery = self.inner.recovery.lock().await;

101 recovery.last_pass = Some(Instant::now());

102 if recovery.generation == generation {

103 recovery.cursor = after;

104 recovery.needed = false;

51105 }

106 Ok(())

52107 }

53108

54109 async fn handle_entry(&self, entry: String, wire: Option<WireMessage>) {

@@ -70,7 +125,7 @@

70125 entry: entry.clone(),

71126 };

72127 if self.inner.events.try_send(event).is_err() {

73 self.inner.transport.inflight.lock().await.remove(&entry);

128 self.release(&entry).await;

74129 }

75130 }

76131 Some(WireMessage::Response { id, result }) => {

@@ -95,7 +150,7 @@

95150 .await,

96151 Ok(Ok(()))

97152 ) {

98 self.inner.transport.inflight.lock().await.remove(&entry);

153 self.release(&entry).await;

99154 }

100155 }

101156 Some(WireMessage::Heartbeat { target }) => {

@@ -106,7 +161,7 @@

106161 {

107162 let _ = self.inner.transport.ack(&entry).await;

108163 } else {

109 self.inner.transport.inflight.lock().await.remove(&entry);

164 self.release(&entry).await;

110165 }

111166 }

112167 None => {

@@ -115,3 +170,8 @@

115170 }

116171 }

117172}

173

174fn backoff(failures: u32) -> Duration {

175 let shift = failures.saturating_sub(1).min(5);

176 (ERROR_BACKOFF * 2u32.pow(shift)).min(MAX_ERROR_BACKOFF)

177}

Modified lib/infra/actor/Cargo.toml +1 −0 View lib/infra/actor/Cargo.toml

@@ -13,4 +13,5 @@

1313serde = { workspace = true, features = ["derive"] }

1414serde_json = { workspace = true }

1515tokio = { workspace = true, features = ["rt", "sync", "time", "macros"] }

16tracing = { workspace = true }

1617uuid = { workspace = true }

Modified Cargo.lock +1 −0 View Cargo.lock

@@ -211,6 +211,7 @@

211211 "serde",

212212 "serde_json",

213213 "tokio",

214 "tracing",

214215 "uuid",

215216]

216217