futu_cache/qot_right/reconnect/
cache.rs

1use super::super::*;
2
3impl QotRightCache {
4    pub fn reconnect_lifecycle_snapshot(&self) -> QotRightReconnectLifecycle {
5        self.meta.lock().reconnect_lifecycle.clone()
6    }
7
8    pub fn register_reconnect_pending(
9        &self,
10        expected_generation: u64,
11        contribution: QotRightReconnectAccumulator,
12    ) -> Option<QotRightReconnectRegistration> {
13        let mut meta = self.meta.lock();
14        if meta.backend_generation != expected_generation {
15            return None;
16        }
17        let lifecycle = std::mem::take(&mut meta.reconnect_lifecycle);
18        let (next, registration) = match lifecycle {
19            QotRightReconnectLifecycle::Idle => {
20                let token = next_reconnect_token(&mut meta);
21                (
22                    QotRightReconnectLifecycle::Pending {
23                        token,
24                        generation: expected_generation,
25                        accumulator: contribution,
26                    },
27                    QotRightReconnectRegistration {
28                        token,
29                        owns_deferred: false,
30                        should_disconnect: true,
31                    },
32                )
33            }
34            QotRightReconnectLifecycle::Deferred {
35                token,
36                generation,
37                mut accumulator,
38                dispatch_admitted,
39            } if generation == expected_generation => {
40                accumulator.merge(&contribution);
41                if dispatch_admitted {
42                    (
43                        QotRightReconnectLifecycle::Deferred {
44                            token,
45                            generation,
46                            accumulator,
47                            dispatch_admitted: true,
48                        },
49                        QotRightReconnectRegistration {
50                            token,
51                            owns_deferred: true,
52                            should_disconnect: false,
53                        },
54                    )
55                } else {
56                    (
57                        QotRightReconnectLifecycle::Pending {
58                            token,
59                            generation,
60                            accumulator,
61                        },
62                        QotRightReconnectRegistration {
63                            token,
64                            owns_deferred: false,
65                            should_disconnect: true,
66                        },
67                    )
68                }
69            }
70            QotRightReconnectLifecycle::Pending {
71                token,
72                generation,
73                mut accumulator,
74            } if generation == expected_generation => {
75                accumulator.merge(&contribution);
76                (
77                    QotRightReconnectLifecycle::Pending {
78                        token,
79                        generation,
80                        accumulator,
81                    },
82                    QotRightReconnectRegistration {
83                        token,
84                        owns_deferred: false,
85                        should_disconnect: false,
86                    },
87                )
88            }
89            QotRightReconnectLifecycle::InFlight {
90                token,
91                generation,
92                serving,
93                mut follow_up,
94            } if generation == expected_generation => {
95                follow_up.merge(&contribution);
96                (
97                    QotRightReconnectLifecycle::InFlight {
98                        token,
99                        generation,
100                        serving,
101                        follow_up,
102                    },
103                    QotRightReconnectRegistration {
104                        token,
105                        owns_deferred: false,
106                        should_disconnect: false,
107                    },
108                )
109            }
110            other => {
111                meta.reconnect_lifecycle = other;
112                return None;
113            }
114        };
115        meta.reconnect_lifecycle = next;
116        Some(registration)
117    }
118
119    pub fn reserve_reconnect_deferred(
120        &self,
121        expected_generation: u64,
122        contribution: QotRightReconnectAccumulator,
123    ) -> Option<QotRightReconnectRegistration> {
124        let mut meta = self.meta.lock();
125        if meta.backend_generation != expected_generation {
126            return None;
127        }
128        let lifecycle = std::mem::take(&mut meta.reconnect_lifecycle);
129        let (next, registration) = match lifecycle {
130            QotRightReconnectLifecycle::Idle => {
131                let token = next_reconnect_token(&mut meta);
132                (
133                    QotRightReconnectLifecycle::Deferred {
134                        token,
135                        generation: expected_generation,
136                        accumulator: contribution,
137                        dispatch_admitted: false,
138                    },
139                    QotRightReconnectRegistration {
140                        token,
141                        owns_deferred: true,
142                        should_disconnect: false,
143                    },
144                )
145            }
146            QotRightReconnectLifecycle::Deferred {
147                token,
148                generation,
149                mut accumulator,
150                dispatch_admitted,
151            } if generation == expected_generation => {
152                accumulator.merge(&contribution);
153                (
154                    QotRightReconnectLifecycle::Deferred {
155                        token,
156                        generation,
157                        accumulator,
158                        dispatch_admitted,
159                    },
160                    QotRightReconnectRegistration {
161                        token,
162                        owns_deferred: true,
163                        should_disconnect: false,
164                    },
165                )
166            }
167            QotRightReconnectLifecycle::Pending {
168                token,
169                generation,
170                mut accumulator,
171            } if generation == expected_generation => {
172                accumulator.merge(&contribution);
173                (
174                    QotRightReconnectLifecycle::Pending {
175                        token,
176                        generation,
177                        accumulator,
178                    },
179                    QotRightReconnectRegistration {
180                        token,
181                        owns_deferred: false,
182                        should_disconnect: false,
183                    },
184                )
185            }
186            QotRightReconnectLifecycle::InFlight {
187                token,
188                generation,
189                serving,
190                mut follow_up,
191            } if generation == expected_generation => {
192                follow_up.merge(&contribution);
193                (
194                    QotRightReconnectLifecycle::InFlight {
195                        token,
196                        generation,
197                        serving,
198                        follow_up,
199                    },
200                    QotRightReconnectRegistration {
201                        token,
202                        owns_deferred: false,
203                        should_disconnect: false,
204                    },
205                )
206            }
207            other => {
208                meta.reconnect_lifecycle = other;
209                return None;
210            }
211        };
212        meta.reconnect_lifecycle = next;
213        Some(registration)
214    }
215
216    pub fn deferred_reconnect_is_active(
217        &self,
218        token: QotRightReconnectToken,
219        generation: u64,
220    ) -> bool {
221        let meta = self.meta.lock();
222        meta.backend_generation == generation
223            && matches!(
224                meta.reconnect_lifecycle,
225                QotRightReconnectLifecycle::Deferred {
226                    token: active_token,
227                    generation: active_generation,
228                    ..
229                } if active_token == token && active_generation == generation
230            )
231    }
232
233    /// Linearize a Subscribe CMD6211 writer admission against entitlement
234    /// pushes. A push that moved the exact owner to `Pending` before this call
235    /// wins and rejects the send. Once the first CMD6211 command is admitted,
236    /// later same-generation pushes merge into the deferred owner without
237    /// disconnecting the route before its ACK transaction completes.
238    pub fn admit_deferred_reconnect_dispatch(
239        &self,
240        token: QotRightReconnectToken,
241        generation: u64,
242    ) -> bool {
243        let mut meta = self.meta.lock();
244        if meta.backend_generation != generation {
245            return false;
246        }
247        let QotRightReconnectLifecycle::Deferred {
248            token: active_token,
249            generation: active_generation,
250            dispatch_admitted,
251            ..
252        } = &mut meta.reconnect_lifecycle
253        else {
254            return false;
255        };
256        if *active_token != token || *active_generation != generation {
257            return false;
258        }
259        *dispatch_admitted = true;
260        true
261    }
262
263    pub fn reconnect_claim_is_active(
264        &self,
265        token: QotRightReconnectToken,
266        generation: u64,
267    ) -> bool {
268        let meta = self.meta.lock();
269        meta.backend_generation == generation
270            && matches!(
271                meta.reconnect_lifecycle,
272                QotRightReconnectLifecycle::InFlight {
273                    token: active_token,
274                    generation: active_generation,
275                    ..
276                } if active_token == token && active_generation == generation
277            )
278    }
279
280    pub fn pending_reconnect_is_active(
281        &self,
282        token: QotRightReconnectToken,
283        generation: u64,
284    ) -> bool {
285        let meta = self.meta.lock();
286        meta.backend_generation == generation
287            && matches!(
288                meta.reconnect_lifecycle,
289                QotRightReconnectLifecycle::Pending {
290                    token: active_token,
291                    generation: active_generation,
292                    ..
293                } if active_token == token && active_generation == generation
294            )
295    }
296
297    /// Freeze the active reconnect accumulator at request-send time. Delayed
298    /// entitlement follow-ups absorb every contribution received during the
299    /// 1s C++ spacing window into the serving request; contributions arriving
300    /// after this transaction remain in `follow_up` for the next generation.
301    pub fn freeze_reconnect_request_claim(
302        &self,
303        token: QotRightReconnectToken,
304        generation: u64,
305    ) -> Option<QotRightReconnectClaim> {
306        let mut meta = self.meta.lock();
307        if meta.backend_generation != generation {
308            return None;
309        }
310        let lifecycle = std::mem::take(&mut meta.reconnect_lifecycle);
311        let QotRightReconnectLifecycle::InFlight {
312            token: active_token,
313            generation: active_generation,
314            mut serving,
315            follow_up,
316        } = lifecycle
317        else {
318            meta.reconnect_lifecycle = lifecycle;
319            return None;
320        };
321        if active_token != token || active_generation != generation {
322            meta.reconnect_lifecycle = QotRightReconnectLifecycle::InFlight {
323                token: active_token,
324                generation: active_generation,
325                serving,
326                follow_up,
327            };
328            return None;
329        }
330        serving.merge(&follow_up);
331        let kind = if serving.is_empty() {
332            QotRightReconnectClaimKind::OrdinaryHighest
333        } else {
334            QotRightReconnectClaimKind::Entitlement(serving.clone())
335        };
336        meta.reconnect_lifecycle = QotRightReconnectLifecycle::InFlight {
337            token,
338            generation,
339            serving,
340            follow_up: QotRightReconnectAccumulator::default(),
341        };
342        Some(QotRightReconnectClaim {
343            token,
344            source_generation: generation.saturating_sub(1),
345            active_generation: generation,
346            kind,
347        })
348    }
349
350    pub fn commit_deferred_reconnect_after_ack<T>(
351        &self,
352        token: QotRightReconnectToken,
353        generation: u64,
354        commit: impl FnOnce() -> T,
355    ) -> Option<T> {
356        let mut meta = self.meta.lock();
357        if meta.backend_generation != generation {
358            return None;
359        }
360        let QotRightReconnectLifecycle::Deferred {
361            token: active_token,
362            generation: active_generation,
363            accumulator,
364            dispatch_admitted,
365        } = &meta.reconnect_lifecycle
366        else {
367            return None;
368        };
369        if *active_token != token || *active_generation != generation || !*dispatch_admitted {
370            return None;
371        }
372        let accumulator = accumulator.clone();
373        let result = commit();
374        meta.reconnect_lifecycle = QotRightReconnectLifecycle::Pending {
375            token,
376            generation,
377            accumulator,
378        };
379        Some(result)
380    }
381
382    pub fn fail_deferred_reconnect(&self, token: QotRightReconnectToken, generation: u64) -> bool {
383        self.commit_deferred_reconnect_after_ack(token, generation, || ())
384            .is_some()
385    }
386
387    /// Freeze the exact source-generation reconnect owner before any socket
388    /// retry begins. An ordinary transport failure creates an empty Pending
389    /// owner; late same-generation entitlement pushes merge into that owner.
390    /// The successor connection must present this token when claiming the
391    /// handoff, so an unrelated reconnect loop cannot consume it.
392    pub fn begin_reconnect_handoff(&self) -> QotRightReconnectHandoff {
393        let mut meta = self.meta.lock();
394        let source_generation = meta.backend_generation;
395        let lifecycle = std::mem::take(&mut meta.reconnect_lifecycle);
396        let (token, next) = match lifecycle {
397            QotRightReconnectLifecycle::Idle => {
398                let token = next_reconnect_token(&mut meta);
399                (
400                    token,
401                    QotRightReconnectLifecycle::Pending {
402                        token,
403                        generation: source_generation,
404                        accumulator: QotRightReconnectAccumulator::default(),
405                    },
406                )
407            }
408            QotRightReconnectLifecycle::Deferred {
409                token,
410                generation,
411                accumulator,
412                ..
413            } if generation == source_generation => (
414                token,
415                QotRightReconnectLifecycle::Pending {
416                    token,
417                    generation,
418                    accumulator,
419                },
420            ),
421            QotRightReconnectLifecycle::Pending {
422                token,
423                generation,
424                accumulator,
425            } if generation == source_generation => (
426                token,
427                QotRightReconnectLifecycle::Pending {
428                    token,
429                    generation,
430                    accumulator,
431                },
432            ),
433            QotRightReconnectLifecycle::InFlight {
434                token,
435                generation,
436                serving,
437                follow_up,
438            } if generation == source_generation => (
439                token,
440                QotRightReconnectLifecycle::InFlight {
441                    token,
442                    generation,
443                    serving,
444                    follow_up,
445                },
446            ),
447            other => {
448                meta.reconnect_lifecycle = other;
449                let token = next_reconnect_token(&mut meta);
450                meta.reconnect_lifecycle = QotRightReconnectLifecycle::Pending {
451                    token,
452                    generation: source_generation,
453                    accumulator: QotRightReconnectAccumulator::default(),
454                };
455                return QotRightReconnectHandoff {
456                    token,
457                    source_generation,
458                };
459            }
460        };
461        meta.reconnect_lifecycle = next;
462        QotRightReconnectHandoff {
463            token,
464            source_generation,
465        }
466    }
467
468    pub fn advance_generation_and_claim_reconnect(
469        &self,
470        handoff: QotRightReconnectHandoff,
471        active_generation: u64,
472    ) -> Option<QotRightReconnectClaim> {
473        let mut meta = self.meta.lock();
474        let source_generation = handoff.source_generation;
475        if meta.backend_generation != source_generation
476            || active_generation != source_generation.saturating_add(1)
477        {
478            return None;
479        }
480        let lifecycle = std::mem::take(&mut meta.reconnect_lifecycle);
481        let (token, serving, kind) = match lifecycle {
482            QotRightReconnectLifecycle::Idle => {
483                let token = next_reconnect_token(&mut meta);
484                (
485                    token,
486                    QotRightReconnectAccumulator::default(),
487                    QotRightReconnectClaimKind::OrdinaryHighest,
488                )
489            }
490            QotRightReconnectLifecycle::Deferred {
491                token,
492                generation,
493                accumulator,
494                ..
495            }
496            | QotRightReconnectLifecycle::Pending {
497                token,
498                generation,
499                accumulator,
500            } if generation == source_generation && token == handoff.token => {
501                let kind = if accumulator.is_empty() {
502                    QotRightReconnectClaimKind::OrdinaryHighest
503                } else {
504                    QotRightReconnectClaimKind::Entitlement(accumulator.clone())
505                };
506                (token, accumulator, kind)
507            }
508            QotRightReconnectLifecycle::InFlight {
509                token,
510                generation,
511                mut serving,
512                follow_up,
513            } if generation == source_generation && token == handoff.token => {
514                serving.merge(&follow_up);
515                let kind = QotRightReconnectClaimKind::Entitlement(serving.clone());
516                (token, serving, kind)
517            }
518            other => {
519                meta.reconnect_lifecycle = other;
520                return None;
521            }
522        };
523        meta.backend_generation = active_generation;
524        meta.freshness = qot_right_on_epoch_advanced(meta.freshness.clone());
525        meta.reconnect_lifecycle = QotRightReconnectLifecycle::InFlight {
526            token,
527            generation: active_generation,
528            serving,
529            follow_up: QotRightReconnectAccumulator::default(),
530        };
531        Some(QotRightReconnectClaim {
532            token,
533            source_generation,
534            active_generation,
535            kind,
536        })
537    }
538
539    pub fn finish_reconnect(
540        &self,
541        token: QotRightReconnectToken,
542        generation: u64,
543    ) -> Option<QotRightReconnectRegistration> {
544        let mut meta = self.meta.lock();
545        if meta.backend_generation != generation {
546            return None;
547        }
548        let lifecycle = std::mem::take(&mut meta.reconnect_lifecycle);
549        let QotRightReconnectLifecycle::InFlight {
550            token: active_token,
551            generation: active_generation,
552            serving,
553            follow_up,
554        } = lifecycle
555        else {
556            meta.reconnect_lifecycle = lifecycle;
557            return None;
558        };
559        if active_token != token || active_generation != generation {
560            meta.reconnect_lifecycle = QotRightReconnectLifecycle::InFlight {
561                token: active_token,
562                generation: active_generation,
563                serving,
564                follow_up,
565            };
566            return None;
567        }
568        if follow_up.is_empty() {
569            meta.reconnect_lifecycle = QotRightReconnectLifecycle::Idle;
570            return Some(QotRightReconnectRegistration {
571                token,
572                owns_deferred: false,
573                should_disconnect: false,
574            });
575        }
576        let next_token = next_reconnect_token(&mut meta);
577        meta.reconnect_lifecycle = QotRightReconnectLifecycle::Pending {
578            token: next_token,
579            generation,
580            accumulator: follow_up,
581        };
582        Some(QotRightReconnectRegistration {
583            token: next_token,
584            owns_deferred: false,
585            should_disconnect: true,
586        })
587    }
588}
futu_cache/qot_right/reconnect/cache.rs

futu_cache/qot_right/reconnect/
cache.rs
