ALL

For reading and verification only. You may not copy, use, modify or redistribute this code. Full notice on the main page.

tunnel/gated_test.go

SHA-256: dd97e773d11ae815c055f1c85e9e38336bfd29e2eee74f8fab88553931ac3840
269 lines · 9041 bytes · View on GitHub

1package main
2
3import (
4 "strings"
5 "testing"
6 "time"
7)
8
9// The gated thin pipe (gated.go): a device that is gated is served only a little, counted by the node on every status the control plane gives.
10
11func TestGated_DayUsageCountsAdmitsAndRollsAtMidnight(t *testing.T) {
12 now := time.Date(2026, 10, 26, 23, 59, 0, 0, time.UTC)
13 u := &dayUsage{day: now.Unix() / 86400}
14 for i := 0; i < 3; i++ {
15 if ok, why := u.admit(now, 1000, 3); !ok {
16 t.Fatalf("stream %d: %s", i, why)
17 }
18 }
19 if ok, why := u.admit(now, 1000, 3); ok || why != "gated_streams" {
20 t.Fatalf("the fourth stream: %v %q", ok, why)
21 }
22 if got := u.charge(now, 600, 1000); got != 600 {
23 t.Fatalf("%d", got)
24 }
25 if got := u.charge(now, 600, 1000); got != 400 {
26 t.Fatalf("the edge gives %d, want the 400 that are left", got)
27 }
28 if got := u.charge(now, 1, 1000); got != 0 {
29 t.Fatalf("after the edge %d", got)
30 }
31 u2 := &dayUsage{day: now.Unix() / 86400, bytes: 1000}
32 if ok, why := u2.admit(now, 1000, 3); ok || why != "gated_bytes" {
33 t.Fatalf("a stream with no byte budget left: %v %q", ok, why)
34 }
35 next := now.Add(2 * time.Minute) // past midnight UTC
36 if ok, _ := u.admit(next, 1000, 3); !ok {
37 t.Fatal("the new day did not reset the stream count")
38 }
39 if got := u.charge(next, 700, 1000); got != 700 {
40 t.Fatalf("the new day did not reset the bytes: %d", got)
41 }
42}
43
44func TestGated_SweepKeepsWhatIsHeldOrUsedToday(t *testing.T) {
45 fc := newFakeClock()
46 g := newGateTable(newClock(fc.Now, fc.AfterFunc))
47 held := g.acquire("held.example")
48 used := g.acquire("used.example")
49 used.charge(fc.Now(), 10, 1000)
50 g.release(used)
51 idle := g.acquire("idle.example")
52 g.release(idle)
53 g.Sweep()
54 if g.size() != 2 {
55 t.Fatalf("%d records after the sweep, want the held one and the used one", g.size())
56 }
57 fc.Advance(25 * time.Hour)
58 g.Sweep()
59 if g.size() != 1 {
60 t.Fatalf("%d records a day later, want only the held one", g.size())
61 }
62 g.release(held)
63 g.Sweep()
64 if g.size() != 0 {
65 t.Fatalf("%d", g.size())
66 }
67}
68
69func gatedEnv(t *testing.T, budget int64, streams int) *testEnv {
70 return newTestEnv(t, func(c *Config) { c.GatedBytesPerDay, c.GatedStreamsPerDay = budget, streams })
71}
72
73func (e *testEnv) gatedBytes(d *testDevice) int64 {
74 e.srv.gates.mu.Lock()
75 u := e.srv.gates.byHost[d.host]
76 e.srv.gates.mu.Unlock()
77 if u == nil {
78 return 0
79 }
80 u.mu.Lock()
81 defer u.mu.Unlock()
82 return u.bytes
83}
84
85func TestGated_ADeviceStopsAtItsByteBudgetAndTheStreamIsCutThere(t *testing.T) {
86 const budget = 60_000
87 e := gatedEnv(t, budget, 200)
88 d := e.newDevice()
89 e.plan(d).status, e.plan(d).plan, e.plan(d).validFor = statusGated, "trial", 100*time.Hour // (the test moves the node's clock a day)
90 dc := e.connectOK(d)
91 if body, _ := dc.ok["body"].(map[string]any); body["status"] != "gated" {
92 t.Fatalf("the OK says %v", dc.ok)
93 }
94 dc.serve(nil)
95 c, err := e.clientTLS(d)
96 if err != nil {
97 t.Fatal(err)
98 }
99 defer c.Close()
100 if resp, err := postOver(c, d.host, []byte("a thin pipe is enough for a card")); err != nil || string(resp) != "echo:a thin pipe is enough for a card" {
101 t.Fatalf("a small exchange of a gated device: %q %v", resp, err)
102 }
103 if _, err := postOver(c, d.host, make([]byte, 400_000)); err == nil {
104 t.Fatal("a body far past the budget went through")
105 }
106 e.waitIdle()
107 if got := e.gatedBytes(d); got != budget {
108 t.Fatalf("the node counted %d bytes, want exactly the budget %d", got, budget)
109 }
110 up, down := e.metricValue(`all_tunnel_bytes_total{direction="up"}`), e.metricValue(`all_tunnel_bytes_total{direction="down"}`)
111 if up+down != budget {
112 t.Fatalf("%d bytes crossed the node (up %d, down %d), want exactly %d", up+down, up, down, budget)
113 }
114 if e.metricValue(`all_tunnel_gated_cut_total`) != 1 || e.metricValue(`all_tunnel_stream_closed_total{reason="gated_bytes"}`) != 1 {
115 t.Fatalf("metrics:\n%s", e.srv.renderMetrics())
116 }
117 if e.logs.Count(`"reason":"gated_bytes"`) < 1 {
118 t.Fatalf("logs:\n%s", e.logs.String())
119 }
120 // a new stream the same day gets nothing: the byte budget is spent
121 if _, err := e.clientTLS(d); err == nil {
122 t.Fatal("a stream was admitted with the day's bytes spent")
123 }
124 if e.rejected("gated_bytes") != 1 {
125 t.Fatalf("counted %d", e.rejected("gated_bytes"))
126 }
127 // reconnecting does not buy a new budget
128 dc.c.Close()
129 dc2 := e.connectOK(d)
130 dc2.serve(nil)
131 if _, err := e.clientTLS(d); err == nil {
132 t.Fatal("a reconnected device got a fresh budget")
133 }
134 // midnight does
135 e.offsetNS.Add(int64(24 * time.Hour))
136 c3, err := e.clientTLS(d)
137 if err != nil {
138 t.Fatalf("the next day: %v", err)
139 }
140 defer c3.Close()
141 if resp, err := postOver(c3, d.host, []byte("tomorrow")); err != nil || string(resp) != "echo:tomorrow" {
142 t.Fatalf("%q %v", resp, err)
143 }
144 e.checkTranscript()
145}
146
147func TestGated_AtMostTheDaysNumberOfStreams(t *testing.T) {
148 e := gatedEnv(t, 50<<20, 3)
149 d := e.newDevice()
150 e.plan(d).status, e.plan(d).plan = statusGated, "trial"
151 dc := e.connectOK(d)
152 dc.serve(nil)
153 for i := 0; i < 3; i++ {
154 c, err := e.clientTLS(d)
155 if err != nil {
156 t.Fatalf("stream %d: %v", i, err)
157 }
158 if _, err := postOver(c, d.host, []byte("x")); err != nil {
159 t.Fatal(err)
160 }
161 c.Close()
162 }
163 if _, err := e.clientTLS(d); err == nil || !strings.Contains(err.Error(), "internal error") {
164 t.Fatalf("the fourth stream of the day: %v", err)
165 }
166 if e.rejected("gated_streams") != 1 {
167 t.Fatalf("counted %d", e.rejected("gated_streams"))
168 }
169 // a device that is not gated is not counted at all
170 other := e.newDevice()
171 oc := e.connectOK(other)
172 oc.serve(nil)
173 for i := 0; i < 6; i++ {
174 c, err := e.clientTLS(other)
175 if err != nil {
176 t.Fatalf("an active device, stream %d: %v", i, err)
177 }
178 c.Close()
179 }
180}
181
182func TestGated_ADeviceThatBecomesGatedAtARenewIsHeldToTheBudgetOnTheStreamsAlreadyOpen(t *testing.T) {
183 const budget = 80_000
184 e := gatedEnv(t, budget, 200)
185 d := e.newDevice()
186 dc := e.connectOK(d) // active, paid: the trial is still on
187 dc.serve(nil)
188 c, err := e.clientTLS(d)
189 if err != nil {
190 t.Fatal(err)
191 }
192 defer c.Close()
193 // far more than the budget goes through while the device is active
194 for i := 0; i < 3; i++ {
195 if resp, err := postOver(c, d.host, make([]byte, 100_000)); err != nil || len(resp) != 100_005 {
196 t.Fatalf("an active device's exchange %d: %d bytes, %v", i, len(resp), err)
197 }
198 }
199 if e.gatedBytes(d) != 0 {
200 t.Fatalf("an active device was counted: %d", e.gatedBytes(d))
201 }
202 // the trial ends: the control plane says gated at the next RENEW, and the node holds the open stream to the budget from there on
203 e.plan(d).status, e.plan(d).plan = statusGated, "trial"
204 dc.drainFeed()
205 if r := dc.renew(); !strings.HasPrefix(r, "OK ") || !strings.Contains(r, `"status":"gated"`) {
206 t.Fatalf("renew: %q", r)
207 }
208 if resp, err := postOver(c, d.host, []byte("a card")); err != nil || string(resp) != "echo:a card" {
209 t.Fatalf("a small exchange after the RENEW: %q %v", resp, err)
210 }
211 if _, err := postOver(c, d.host, make([]byte, 300_000)); err == nil {
212 t.Fatal("the stream that was already open went past the budget after the device became gated")
213 }
214 e.waitIdle()
215 if got := e.gatedBytes(d); got != budget {
216 t.Fatalf("counted %d, want exactly %d", got, budget)
217 }
218 // and back to active at the next RENEW: nothing is counted or cut any more, what was used stays on the day's record
219 e.plan(d).status, e.plan(d).plan = statusActive, "paid"
220 if r := dc.renew(); !strings.HasPrefix(r, "OK ") || !strings.Contains(r, `"status":"active"`) {
221 t.Fatalf("renew: %q", r)
222 }
223 c2, err := e.clientTLS(d)
224 if err != nil {
225 t.Fatal(err)
226 }
227 defer c2.Close()
228 if resp, err := postOver(c2, d.host, make([]byte, 200_000)); err != nil || len(resp) != 200_005 {
229 t.Fatalf("a device that is active again: %d bytes, %v", len(resp), err)
230 }
231 if e.gatedBytes(d) != budget {
232 t.Fatalf("an active device was counted again: %d", e.gatedBytes(d))
233 }
234 e.checkTranscript()
235}
236
237func TestGated_AFeedRefreshThatMakesTheHostGatedAlsoHoldsItsStreams(t *testing.T) {
238 const budget = 70_000
239 e, cp := httpNode(t, nil, func(c *Config) { c.GatedBytesPerDay = budget })
240 cp.mu.Lock()
241 cp.hold = 10 * time.Second
242 cp.mu.Unlock()
243 d := e.newDevice()
244 dc := e.connectOK(d)
245 dc.serve(nil)
246 c, err := e.clientTLS(d)
247 if err != nil {
248 t.Fatal(err)
249 }
250 defer c.Close()
251 if _, err := postOver(c, d.host, make([]byte, 100_000)); err != nil {
252 t.Fatalf("while active: %v", err)
253 }
254 // a resume event after the control plane turned the device gated (the node asks authorize again, outside any RENEW of the device)
255 e.plan(d).status, e.plan(d).plan = statusGated, "trial"
256 cp.addEvent("resume", d)
257 eventually(t, "the refresh asked authorize again", 5*time.Second, func() bool { return e.cp.callsFor(d.id, "renew") >= 1 })
258 eventually(t, "the host is gated", 5*time.Second, func() bool {
259 cc, he := e.srv.routes.lookup(d.host)
260 return cc != nil && he.gated.Load()
261 })
262 if _, err := postOver(c, d.host, make([]byte, 300_000)); err == nil {
263 t.Fatal("the stream was not held to the budget after the refresh")
264 }
265 e.waitIdle()
266 if got := e.gatedBytes(d); got != budget {
267 t.Fatalf("counted %d, want %d", got, budget)
268 }
269}