route_retirement.go RawHistoryFolder

1package daemon
2
3import (
4	"errors"
5	"maps"
6	"time"
7
8	"github.com/bnema/vev/internal/protocol"
9)
10
11// retiredRoutes uses only authoritative observations: this daemon's own
12// lifecycles. Another daemon's routes are the client's broker to retire.
13func retiredRoutes(subscription protocol.RouteAttentionSubscription, _ map[protocol.RouteAttentionTarget]time.Time, inv sessionInventory) []protocol.RouteRetired {
14	var retired []protocol.RouteRetired
15	for _, target := range subscription.Targets {
16		if target.SourceKey != "" {
17			continue
18		}
19		present := false
20		for _, item := range inv.live {
21			present = present || item.view.incarnation == target.Target.LifecycleID
22		}
23		for _, stopped := range inv.stopped {
24			present = present || stopped.incarnation == target.Target.LifecycleID
25		}
26		if !present {
27			retired = append(retired, protocol.RouteRetired{Ref: target.Ref, Target: target.Target})
28		}
29	}
30	return retired
31}
32
33func (d *Daemon) reconcileRouteHistory(ac *attachedClient, admitted ...*attachmentEffect) {
34	if ac == nil {
35		return
36	}
37	ac.routeMu.RLock()
38	expected := ac.routeSubscriptionTransport
39	subscription := ac.routeAttentionSubscription
40	subscription.Targets = append([]protocol.RouteAttentionTarget(nil), subscription.Targets...)
41	after := maps.Clone(ac.routeObservationAfter)
42	ac.routeMu.RUnlock()
43	if expected.transport == nil || len(subscription.Targets) == 0 || !ac.transportSnapshotCurrent(expected) {
44		return
45	}
46	var effect *attachmentEffect
47	if len(admitted) != 0 {
48		effect = admitted[0]
49	} else {
50		_, token, ok := d.currentAttachmentConnection(ac, expected.transport)
51		if !ok {
52			return
53		}
54		var okEffect bool
55		effect, okEffect = ac.beginAttachmentEffect(token)
56		if !okEffect {
57			return
58		}
59		defer effect.End()
60	}
61	if effect == nil || effect.transport != expected || !effect.current() {
62		return
63	}
64	retired := retiredRoutes(subscription, after, d.captureSessionInventory(viewOptions{}, false))
65	for _, message := range retired {
66		// Bind to the connection that supplied this subscription; a resumed
67		// attachment must not receive observations from its old transport.
68		_, err := d.boundedSendWith(expected.transport, func() error {
69			return effect.sendControl(message)
70		})
71		if err != nil {
72			if errors.Is(err, errSendTimedOut) {
73				_ = ac.closeCapturedTransport(expected.transport)
74			}
75			return
76		}
77	}
78}
79
80func (d *Daemon) reconcileAllRouteHistories() {
81	d.mu.Lock()
82	sessions := d.sessionsSnapshotLocked()
83	d.mu.Unlock()
84	for _, sess := range sessions {
85		for _, ac := range sess.snapshotAttachments() {
86			d.reconcileRouteHistory(ac)
87		}
88	}
89}