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    /// Promote a refresh-created reservation when the enclosing subscription
388    /// request fails, whether backend dispatch was admitted or not. The rights
389    /// change still requires the same generation-fenced reconnect; leaving the
390    /// owner in `Deferred` would strand that reconnect forever.
391    pub fn promote_deferred_reconnect_for_request_failure(
392        &self,
393        token: QotRightReconnectToken,
394        generation: u64,
395    ) -> bool {
396        let mut meta = self.meta.lock();
397        if meta.backend_generation != generation {
398            return false;
399        }
400        let QotRightReconnectLifecycle::Deferred {
401            token: active_token,
402            generation: active_generation,
403            accumulator,
404            ..
405        } = &meta.reconnect_lifecycle
406        else {
407            return false;
408        };
409        if *active_token != token || *active_generation != generation {
410            return false;
411        }
412        let accumulator = accumulator.clone();
413        meta.reconnect_lifecycle = QotRightReconnectLifecycle::Pending {
414            token,
415            generation,
416            accumulator,
417        };
418        true
419    }
420
421    /// Freeze the exact source-generation reconnect owner before any socket
422    /// retry begins. An ordinary transport failure creates an empty Pending
423    /// owner; late same-generation entitlement pushes merge into that owner.
424    /// The successor connection must present this token when claiming the
425    /// handoff, so an unrelated reconnect loop cannot consume it.
426    pub fn begin_reconnect_handoff(&self) -> QotRightReconnectHandoff {
427        let mut meta = self.meta.lock();
428        let source_generation = meta.backend_generation;
429        let lifecycle = std::mem::take(&mut meta.reconnect_lifecycle);
430        let (token, next) = match lifecycle {
431            QotRightReconnectLifecycle::Idle => {
432                let token = next_reconnect_token(&mut meta);
433                (
434                    token,
435                    QotRightReconnectLifecycle::Pending {
436                        token,
437                        generation: source_generation,
438                        accumulator: QotRightReconnectAccumulator::default(),
439                    },
440                )
441            }
442            QotRightReconnectLifecycle::Deferred {
443                token,
444                generation,
445                accumulator,
446                ..
447            } if generation == source_generation => (
448                token,
449                QotRightReconnectLifecycle::Pending {
450                    token,
451                    generation,
452                    accumulator,
453                },
454            ),
455            QotRightReconnectLifecycle::Pending {
456                token,
457                generation,
458                accumulator,
459            } if generation == source_generation => (
460                token,
461                QotRightReconnectLifecycle::Pending {
462                    token,
463                    generation,
464                    accumulator,
465                },
466            ),
467            QotRightReconnectLifecycle::InFlight {
468                token,
469                generation,
470                serving,
471                follow_up,
472            } if generation == source_generation => (
473                token,
474                QotRightReconnectLifecycle::InFlight {
475                    token,
476                    generation,
477                    serving,
478                    follow_up,
479                },
480            ),
481            other => {
482                meta.reconnect_lifecycle = other;
483                let token = next_reconnect_token(&mut meta);
484                meta.reconnect_lifecycle = QotRightReconnectLifecycle::Pending {
485                    token,
486                    generation: source_generation,
487                    accumulator: QotRightReconnectAccumulator::default(),
488                };
489                return QotRightReconnectHandoff {
490                    token,
491                    source_generation,
492                };
493            }
494        };
495        meta.reconnect_lifecycle = next;
496        QotRightReconnectHandoff {
497            token,
498            source_generation,
499        }
500    }
501
502    pub fn advance_generation_and_claim_reconnect(
503        &self,
504        handoff: QotRightReconnectHandoff,
505        active_generation: u64,
506    ) -> Option<QotRightReconnectClaim> {
507        let mut meta = self.meta.lock();
508        let source_generation = handoff.source_generation;
509        if meta.backend_generation != source_generation
510            || active_generation != source_generation.saturating_add(1)
511        {
512            return None;
513        }
514        let lifecycle = std::mem::take(&mut meta.reconnect_lifecycle);
515        let (token, serving, kind) = match lifecycle {
516            QotRightReconnectLifecycle::Idle => {
517                let token = next_reconnect_token(&mut meta);
518                (
519                    token,
520                    QotRightReconnectAccumulator::default(),
521                    QotRightReconnectClaimKind::OrdinaryHighest,
522                )
523            }
524            QotRightReconnectLifecycle::Deferred {
525                token,
526                generation,
527                accumulator,
528                ..
529            }
530            | QotRightReconnectLifecycle::Pending {
531                token,
532                generation,
533                accumulator,
534            } if generation == source_generation && token == handoff.token => {
535                let kind = if accumulator.is_empty() {
536                    QotRightReconnectClaimKind::OrdinaryHighest
537                } else {
538                    QotRightReconnectClaimKind::Entitlement(accumulator.clone())
539                };
540                (token, accumulator, kind)
541            }
542            QotRightReconnectLifecycle::InFlight {
543                token,
544                generation,
545                mut serving,
546                follow_up,
547            } if generation == source_generation && token == handoff.token => {
548                serving.merge(&follow_up);
549                let kind = QotRightReconnectClaimKind::Entitlement(serving.clone());
550                (token, serving, kind)
551            }
552            other => {
553                meta.reconnect_lifecycle = other;
554                return None;
555            }
556        };
557        meta.backend_generation = active_generation;
558        meta.source_publication_epoch = None;
559        meta.freshness = qot_right_on_epoch_advanced(meta.freshness.clone());
560        meta.reconnect_lifecycle = QotRightReconnectLifecycle::InFlight {
561            token,
562            generation: active_generation,
563            serving,
564            follow_up: QotRightReconnectAccumulator::default(),
565        };
566        Some(QotRightReconnectClaim {
567            token,
568            source_generation,
569            active_generation,
570            kind,
571        })
572    }
573
574    pub fn finish_reconnect(
575        &self,
576        token: QotRightReconnectToken,
577        generation: u64,
578    ) -> Option<QotRightReconnectRegistration> {
579        let mut meta = self.meta.lock();
580        if meta.backend_generation != generation {
581            return None;
582        }
583        let lifecycle = std::mem::take(&mut meta.reconnect_lifecycle);
584        let QotRightReconnectLifecycle::InFlight {
585            token: active_token,
586            generation: active_generation,
587            serving,
588            follow_up,
589        } = lifecycle
590        else {
591            meta.reconnect_lifecycle = lifecycle;
592            return None;
593        };
594        if active_token != token || active_generation != generation {
595            meta.reconnect_lifecycle = QotRightReconnectLifecycle::InFlight {
596                token: active_token,
597                generation: active_generation,
598                serving,
599                follow_up,
600            };
601            return None;
602        }
603        if follow_up.is_empty() {
604            meta.reconnect_lifecycle = QotRightReconnectLifecycle::Idle;
605            return Some(QotRightReconnectRegistration {
606                token,
607                owns_deferred: false,
608                should_disconnect: false,
609            });
610        }
611        let next_token = next_reconnect_token(&mut meta);
612        meta.reconnect_lifecycle = QotRightReconnectLifecycle::Pending {
613            token: next_token,
614            generation,
615            accumulator: follow_up,
616        };
617        Some(QotRightReconnectRegistration {
618            token: next_token,
619            owns_deferred: false,
620            should_disconnect: true,
621        })
622    }
623}
futu_cache/qot_right/reconnect/cache.rs

futu_cache/qot_right/reconnect/
cache.rs
