broker_stream_window.go RawHistoryFolder

1package ports
2
3import (
4	"sync"
5)
6
7// Logical stream admission window.
8//
9// One connection's logical stream IDs are allocated by its calling service and
10// are strictly increasing, but two concurrent opens may be scheduled in either
11// order: an ID can be abandoned, and a lower ID can legitimately arrive after a
12// higher one was already admitted. A plain "not above the last admitted ID"
13// check would refuse that valid concurrency, so every admission layer (the
14// client, the listener, and the pool) applies one bounded anti-replay window
15// instead.
16
17const (
18	// BrokerStreamWindowSize is the anti-replay window: an identity that trails
19	// the newest admitted identity by BrokerStreamWindowSize or more is stale
20	// and never admitted, while every unconsumed identity inside the window is
21	// admitted exactly once.
22	BrokerStreamWindowSize = 1024
23	// brokerStreamWindowWords is the window's bitset size in 64-bit words.
24	brokerStreamWindowWords = BrokerStreamWindowSize / 64
25)
26
27// BrokerStreamWindow is the bounded anti-replay admission window for the
28// logical stream IDs of exactly one connection scope. Every identity is
29// consumed at most once: a duplicate inside the window, and an identity evicted
30// from it, are both refused as stale. It is safe for concurrent use and
31// performs no I/O.
32//
33// The window is deliberately transport- and codec-independent: the broker IPC
34// client, the broker IPC listener, and the broker core pool share this exact
35// rule, so a request refused by one layer is refused by the next for the same
36// reason rather than by a weaker monotone comparison.
37type BrokerStreamWindow struct {
38	mu      sync.Mutex
39	highest BrokerStreamID
40	// base is the lowest identity the window still represents. It is zero
41	// before the first admission and at least one afterwards.
42	base  BrokerStreamID
43	words [brokerStreamWindowWords]uint64
44}
45
46// Admit consumes one stream identity for this connection. It reports
47// BrokerAdmissionInvalid for a zero identity, and BrokerAdmissionStale for an
48// identity already consumed or evicted from the window because it trails the
49// newest admitted identity by BrokerStreamWindowSize or more. An admitted
50// identity stays consumed even when the caller refuses the open afterwards, so
51// a rejected request can never be replayed with the same identity.
52func (w *BrokerStreamWindow) Admit(stream BrokerStreamID) error {
53	if w == nil {
54		return BrokerAdmissionInvalid
55	}
56	if stream == 0 {
57		return BrokerAdmissionInvalid
58	}
59	w.mu.Lock()
60	defer w.mu.Unlock()
61	if w.highest == 0 {
62		// The first admission opens a window anchored at one, so a lower
63		// identity that a concurrent open legitimately allocated earlier can
64		// still arrive afterwards.
65		w.highest = stream
66		w.base = windowBase(stream)
67		w.setLocked(stream - w.base)
68		return nil
69	}
70	if stream < w.base {
71		return BrokerAdmissionStale
72	}
73	if stream <= w.highest {
74		offset := stream - w.base
75		if w.words[offset/64]&(1<<(offset%64)) != 0 {
76			return BrokerAdmissionStale
77		}
78		w.setLocked(offset)
79		return nil
80	}
81	// The incoming identity is the newest: slide the window forward. Only then
82	// can an identity leave the window, which is what makes a very late open
83	// stale instead of merely duplicate.
84	next := windowBase(stream)
85	if next > w.base {
86		w.shiftLocked(next - w.base)
87		w.base = next
88	}
89	w.highest = stream
90	w.setLocked(stream - w.base)
91	return nil
92}
93
94// windowBase returns the lowest identity a window whose newest admitted
95// identity is highest still represents.
96func windowBase(highest BrokerStreamID) BrokerStreamID {
97	if highest <= BrokerStreamWindowSize {
98		return 1
99	}
100	return highest - (BrokerStreamWindowSize - 1)
101}
102
103// setLocked marks one window offset as consumed.
104func (w *BrokerStreamWindow) setLocked(offset BrokerStreamID) {
105	w.words[offset/64] |= 1 << (offset % 64)
106}
107
108// shiftLocked drops the oldest shift identities from the window. A shift at or
109// above the window size clears it entirely.
110func (w *BrokerStreamWindow) shiftLocked(shift BrokerStreamID) {
111	if shift >= BrokerStreamWindowSize {
112		w.words = [brokerStreamWindowWords]uint64{}
113		return
114	}
115	words := int(shift / 64)
116	bits := uint(shift % 64)
117	var shifted [brokerStreamWindowWords]uint64
118	for i := range shifted {
119		index := i + words
120		if index >= brokerStreamWindowWords {
121			break
122		}
123		// Go defines a shift at or above the operand width as zero, so a zero
124		// bit shift needs no special case.
125		var high uint64
126		if index+1 < brokerStreamWindowWords {
127			high = w.words[index+1] << (64 - bits)
128		}
129		shifted[i] = w.words[index]>>bits | high
130	}
131	w.words = shifted
132}