Skip to content

Commit 7d2f71e

Browse files
committed
Log the feed's in-band stream renewal
A server that bounds the ticket stream by its token's expiry advertises the deadline in the welcome, and the SDK renews the socket in band before it. The connector logs every other socket lifecycle callback and none of these three, so a renewal that gave up left only an unexplained remote close. Log the advertised deadline, each renewal, and each failure, saying whether it is retried or left to the server's lease.
1 parent 7e2d875 commit 7d2f71e

2 files changed

Lines changed: 176 additions & 0 deletions

File tree

‎internal/connector/intake.go‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -755,6 +755,25 @@ func (in *Intake) observer(ctx context.Context) eventfeed.Observer {
755755
BufferOverflow: func(dropped int) {
756756
in.log.Warn("live buffer overflowed", "dropped", dropped)
757757
},
758+
// In-band renewal. A server that bounds the stream by its token's
759+
// expiry advertises the deadline in the welcome, and the package
760+
// renews the socket before it. Nothing here decides anything: a
761+
// renewal that gives up leaves the socket to the server's lease, and
762+
// the reconnect after that `remote` close is the ordinary one. The
763+
// lines are what tell that close apart from an unexplained one.
764+
StreamDeadline: func(expiresAt time.Time) {
765+
in.log.Info("feed stream deadline advertised; renewing in band before it", "expires_at", expiresAt.UTC())
766+
},
767+
Renewed: func(expiresAt time.Time) {
768+
in.log.Info("feed stream renewed in band", "expires_at", expiresAt.UTC())
769+
},
770+
RenewFailed: func(kind eventfeed.RenewFailure, retrying bool) {
771+
msg := "feed stream renewal failed; retrying"
772+
if !retrying {
773+
msg = "feed stream renewal gave up; the server closes the socket at its deadline, and the connector reconnects and catches up"
774+
}
775+
in.log.Warn(msg, "kind", kind.String())
776+
},
758777
}
759778
}
760779

Lines changed: 157 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,157 @@
1+
package connector
2+
3+
import (
4+
"bufio"
5+
"context"
6+
"encoding/json"
7+
"log/slog"
8+
"strings"
9+
"testing"
10+
"time"
11+
12+
"github.com/stretchr/testify/assert"
13+
"github.com/stretchr/testify/require"
14+
15+
"github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed"
16+
"github.com/basecamp/basecamp-sdk/go/pkg/basecamp/eventfeed/feedtest"
17+
)
18+
19+
// logLines decodes a JSON log, one record per line.
20+
func logLines(t *testing.T, raw string) []map[string]any {
21+
t.Helper()
22+
var lines []map[string]any
23+
scanner := bufio.NewScanner(strings.NewReader(raw))
24+
for scanner.Scan() {
25+
var line map[string]any
26+
require.NoError(t, json.Unmarshal(scanner.Bytes(), &line), "log line %q", scanner.Text())
27+
lines = append(lines, line)
28+
}
29+
require.NoError(t, scanner.Err())
30+
return lines
31+
}
32+
33+
// logLine returns the one record whose message is msg, or nil.
34+
func logLine(t *testing.T, raw, msg string) map[string]any {
35+
t.Helper()
36+
var found map[string]any
37+
for _, line := range logLines(t, raw) {
38+
if line["msg"] == msg {
39+
require.Nil(t, found, "more than one %q line", msg)
40+
found = line
41+
}
42+
}
43+
return found
44+
}
45+
46+
// writtenCommand returns the identifier of the first write on conn whose
47+
// command is command.
48+
func writtenCommand(conn *feedtest.Conn, command string) (string, bool) {
49+
for _, write := range conn.Writes() {
50+
var c struct {
51+
Command string `json:"command"`
52+
Identifier string `json:"identifier"`
53+
}
54+
if json.Unmarshal(write, &c) == nil && c.Command == command && c.Identifier != "" {
55+
return c.Identifier, true
56+
}
57+
}
58+
return "", false
59+
}
60+
61+
// A server that bounds the stream advertises its deadline in the welcome, and
62+
// the package renews the socket in band before it. Through the real run loop:
63+
// the advert and the renewal each reach the connector's log, with the
64+
// deadlines they carry, and the fresh ticket the renewal sends does not.
65+
func TestAnInBandRenewalReachesTheConnectorLog(t *testing.T) {
66+
var logs safeBuffer
67+
ledger := newTestLedger(t)
68+
intake, transport, minter, polls := newFeedIntake(t, ledger, Options{
69+
Logger: slog.New(slog.NewJSONHandler(&logs, nil)),
70+
})
71+
minter.ScriptTicket(ticket())
72+
const renewalTicket = "renewal-ticket-that-must-stay-out-of-the-log"
73+
minter.ScriptTicket(eventfeed.StreamTicket{Ticket: renewalTicket, ExpiresIn: 120,
74+
URL: "wss://cable.basecamp.com/cable?ticket=" + renewalTicket})
75+
polls.ScriptPage(eventfeed.PollPage{Position: "p1"})
76+
77+
ctx, cancel := context.WithCancel(context.Background())
78+
defer cancel()
79+
done := runInBackground(ctx, t, intake)
80+
81+
var conn *feedtest.Conn
82+
require.Eventually(t, func() bool {
83+
conn = transport.LastConn()
84+
return conn != nil
85+
}, 5*time.Second, 5*time.Millisecond, "the connector should dial")
86+
87+
// Inside the package's renewal lead already, so the renewal is due as soon
88+
// as the socket is streaming: no wait on the wall clock.
89+
advertised := time.Now().Add(2 * time.Minute).UTC().Truncate(time.Millisecond)
90+
conn.Serve(mustJSON(t, map[string]any{"type": "welcome", "stream_expires_at": advertised.Format(time.RFC3339Nano)}))
91+
92+
var identifier string
93+
require.Eventually(t, func() bool {
94+
var ok bool
95+
identifier, ok = writtenCommand(conn, "subscribe")
96+
return ok
97+
}, 5*time.Second, 5*time.Millisecond, "the connector should subscribe")
98+
conn.Serve(confirmFrame(t, identifier))
99+
100+
require.Eventually(t, func() bool {
101+
_, ok := writtenCommand(conn, "renew")
102+
return ok
103+
}, 5*time.Second, 5*time.Millisecond, "a socket streaming inside its renewal lead renews")
104+
105+
renewed := advertised.Add(time.Hour)
106+
conn.Serve(mustJSON(t, map[string]any{"identifier": identifier, "type": "renewed", "stream_expires_at": renewed.Format(time.RFC3339Nano)}))
107+
108+
require.Eventually(t, func() bool {
109+
return strings.Contains(logs.String(), "feed stream renewed in band")
110+
}, 5*time.Second, 5*time.Millisecond, "the renewal reaches the log")
111+
112+
cancel()
113+
awaitReturn(t, done, "Run should return on shutdown")
114+
115+
raw := logs.String()
116+
deadline := logLine(t, raw, "feed stream deadline advertised; renewing in band before it")
117+
require.NotNil(t, deadline, "the welcome's advert reaches the log:\n%s", raw)
118+
assert.Equal(t, "INFO", deadline["level"])
119+
assert.Equal(t, advertised.Format(time.RFC3339Nano), deadline["expires_at"])
120+
121+
renewal := logLine(t, raw, "feed stream renewed in band")
122+
require.NotNil(t, renewal)
123+
assert.Equal(t, "INFO", renewal["level"])
124+
assert.Equal(t, renewed.Format(time.RFC3339Nano), renewal["expires_at"])
125+
126+
for _, line := range logLines(t, raw) {
127+
assert.NotContains(t, line["msg"], "renewal failed", "nothing failed here")
128+
assert.NotContains(t, line["msg"], "renewal gave up", "nothing failed here")
129+
}
130+
assert.NotContains(t, raw, renewalTicket, "the renewal's ticket is a bearer credential")
131+
}
132+
133+
// A failed renewal is retried once, then left to the server's lease, whose
134+
// `remote` close the connector recovers from as from any other. The log says
135+
// which: the first is a retry, and the second explains the close that follows
136+
// it. The kind is the package's closed vocabulary, never the server's reason.
137+
func TestAFailedRenewalSaysWhetherItIsRetriedOrLeftToTheLease(t *testing.T) {
138+
var logs safeBuffer
139+
intake, _, _ := newTestIntake(t, nil, nil)
140+
intake.log = slog.New(slog.NewJSONHandler(&logs, nil))
141+
142+
observer := intake.observer(context.Background())
143+
require.NotNil(t, observer.RenewFailed)
144+
observer.RenewFailed(eventfeed.RenewRefused, true)
145+
observer.RenewFailed(eventfeed.RenewNoReply, false)
146+
147+
lines := logLines(t, logs.String())
148+
require.Len(t, lines, 2)
149+
150+
assert.Equal(t, "WARN", lines[0]["level"])
151+
assert.Equal(t, "feed stream renewal failed; retrying", lines[0]["msg"])
152+
assert.Equal(t, "refused", lines[0]["kind"])
153+
154+
assert.Equal(t, "WARN", lines[1]["level"])
155+
assert.Equal(t, "feed stream renewal gave up; the server closes the socket at its deadline, and the connector reconnects and catches up", lines[1]["msg"])
156+
assert.Equal(t, "no_reply", lines[1]["kind"])
157+
}

0 commit comments

Comments
 (0)