(&mut self)
| 1254 | } |
| 1255 | |
| 1256 | fn recycle(&mut self) -> Result<()> { |
| 1257 | // Refresh state: sync local handshakes to KvStore, update local last_seen from global |
| 1258 | if let Err(err) = self.refresh_state() { |
| 1259 | warn!("failed to refresh state: {err:?}"); |
| 1260 | } |
| 1261 | |
| 1262 | // Note: Gateway nodes are not removed from KvStore, only marked offline/retired |
| 1263 | |
| 1264 | // Recycle stale CVM instances based on global last_seen (max across all nodes) |
| 1265 | let stale_timeout = self.config.recycle.timeout; |
| 1266 | let now = SystemTime::now(); |
| 1267 | |
| 1268 | let stale_instances: Vec<_> = self |
| 1269 | .state |
| 1270 | .instances |
| 1271 | .iter() |
| 1272 | .filter(|(id, info)| { |
| 1273 | // Skip if instance was registered recently |
| 1274 | if info.reg_time.elapsed().unwrap_or_default() <= stale_timeout { |
| 1275 | return false; |
| 1276 | } |
| 1277 | // Check global last_seen from KvStore (max across all nodes) |
| 1278 | let global_ts = self.kv_store.get_instance_latest_handshake(id); |
| 1279 | let last_seen = global_ts.map(decode_ts).unwrap_or(info.reg_time); |
| 1280 | let elapsed = now.duration_since(last_seen).unwrap_or_default(); |
| 1281 | if elapsed > stale_timeout { |
| 1282 | debug!( |
| 1283 | "stale instance: {} last_seen={:?} ({:?} ago)", |
| 1284 | id, last_seen, elapsed |
| 1285 | ); |
| 1286 | true |
| 1287 | } else { |
| 1288 | false |
| 1289 | } |
| 1290 | }) |
| 1291 | .map(|(id, _)| id.clone()) |
| 1292 | .collect(); |
| 1293 | |
| 1294 | let num_recycled = stale_instances.len(); |
| 1295 | for id in stale_instances { |
| 1296 | self.remove_instance(&id)?; |
| 1297 | } |
| 1298 | |
| 1299 | if num_recycled > 0 { |
| 1300 | info!("recycled {num_recycled} stale instances"); |
| 1301 | self.reconfigure()?; |
| 1302 | } |
| 1303 | Ok(()) |
| 1304 | } |
| 1305 | |
| 1306 | pub(crate) fn exit(&mut self) -> ! { |
| 1307 | std::process::exit(0); |
no test coverage detected