Let the config choose protocol 2 or 3, and fall back for one session only.
The default stays v2. A reflector that refuses that version and names the other gets one session of it, then the station announces the configured protocol again. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
+36
-15
@@ -30,11 +30,15 @@ var (
|
||||
// Run keeps the reflector session until ctx is canceled.
|
||||
func (s *Station) Run(ctx context.Context) error {
|
||||
defer s.Close()
|
||||
major := uint16(3)
|
||||
var major uint16
|
||||
holdMajor := false
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
if !holdMajor {
|
||||
major = s.configuredMajor()
|
||||
}
|
||||
s.mu.Lock()
|
||||
if s.fatal != "" || !s.loaded {
|
||||
s.mu.Unlock()
|
||||
@@ -61,6 +65,7 @@ func (s *Station) Run(ctx context.Context) error {
|
||||
stale := s.epoch != epoch
|
||||
s.mu.Unlock()
|
||||
if stale {
|
||||
holdMajor = false
|
||||
if conn != nil {
|
||||
// This dial used the config from before Save.
|
||||
_ = conn.Close()
|
||||
@@ -72,7 +77,6 @@ func (s *Station) Run(ctx context.Context) error {
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
major = 3
|
||||
s.mu.Lock()
|
||||
s.noteDownLocked(s.clock.Now(), true)
|
||||
s.mu.Unlock()
|
||||
@@ -82,7 +86,7 @@ func (s *Station) Run(ctx context.Context) error {
|
||||
continue
|
||||
}
|
||||
|
||||
err = s.session(ctx, conn, major, epoch)
|
||||
err = s.session(ctx, conn, major, epoch, major == s.configuredMajor())
|
||||
conn.Close()
|
||||
fallback := errors.Is(err, errFallback)
|
||||
reconnect := errors.Is(err, errReconnect)
|
||||
@@ -111,9 +115,10 @@ func (s *Station) Run(ctx context.Context) error {
|
||||
return nil
|
||||
}
|
||||
if fallback {
|
||||
major = 2
|
||||
major = otherMajor(major)
|
||||
holdMajor = true
|
||||
} else {
|
||||
major = 3
|
||||
holdMajor = false
|
||||
}
|
||||
continue
|
||||
}
|
||||
@@ -124,7 +129,7 @@ func (s *Station) Run(ctx context.Context) error {
|
||||
}
|
||||
fatal := s.fatal != ""
|
||||
s.mu.Unlock()
|
||||
major = 3
|
||||
holdMajor = false
|
||||
if ctx.Err() != nil {
|
||||
return nil
|
||||
}
|
||||
@@ -297,7 +302,23 @@ const (
|
||||
phChallenge
|
||||
)
|
||||
|
||||
func (s *Station) session(ctx context.Context, conn net.Conn, major uint16, epoch uint64) error {
|
||||
func (s *Station) configuredMajor() uint16 {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.cfg.protocolMajor == 3 {
|
||||
return 3
|
||||
}
|
||||
return 2
|
||||
}
|
||||
|
||||
func otherMajor(major uint16) uint16 {
|
||||
if major == 2 {
|
||||
return 3
|
||||
}
|
||||
return 2
|
||||
}
|
||||
|
||||
func (s *Station) session(ctx context.Context, conn net.Conn, major uint16, epoch uint64, allowFallback bool) error {
|
||||
s.mu.Lock()
|
||||
s.conn = conn
|
||||
stale := s.epoch != epoch
|
||||
@@ -428,7 +449,7 @@ func (s *Station) session(ctx context.Context, conn net.Conn, major uint16, epoc
|
||||
if udpRx.Before(ex.limit) {
|
||||
ex.limit = udpRx
|
||||
}
|
||||
err := s.onFrame(ctx, wire, major, tcpf, ex)
|
||||
err := s.onFrame(ctx, wire, major, tcpf, ex, allowFallback)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -512,14 +533,14 @@ func (s *Station) onClock(now time.Time) {
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Station) onFrame(ctx context.Context, wire *framed, major uint16, f frame, ex *exchange) error {
|
||||
func (s *Station) onFrame(ctx context.Context, wire *framed, major uint16, f frame, ex *exchange, allowFallback bool) error {
|
||||
switch f.typ {
|
||||
case tcpHeartbeat:
|
||||
return nil
|
||||
case tcpError:
|
||||
return s.onReflectorError(f.body, major)
|
||||
return s.onReflectorError(f.body, major, allowFallback)
|
||||
case tcpProtoDown:
|
||||
return s.onDowngrade(f.body, major)
|
||||
return s.onDowngrade(f.body, major, allowFallback)
|
||||
}
|
||||
if !ex.authed && f.typ >= 100 {
|
||||
return s.retry("Protocol error", nil)
|
||||
@@ -530,25 +551,25 @@ func (s *Station) onFrame(ctx context.Context, wire *framed, major uint16, f fra
|
||||
return s.onV3(ctx, wire, f, ex)
|
||||
}
|
||||
|
||||
func (s *Station) onReflectorError(body []byte, major uint16) error {
|
||||
func (s *Station) onReflectorError(body []byte, major uint16, allowFallback bool) error {
|
||||
msg, _, err := parseString(body)
|
||||
if err != nil {
|
||||
s.endFatalCause("could not be read", err, true)
|
||||
return fmt.Errorf("could not be read: %w", err)
|
||||
}
|
||||
if major >= 3 && fallback20(msg) {
|
||||
if allowFallback && wantsOther(major, msg) {
|
||||
return errFallback
|
||||
}
|
||||
s.endFatal(msg, true)
|
||||
return errStopped
|
||||
}
|
||||
|
||||
func (s *Station) onDowngrade(body []byte, major uint16) error {
|
||||
func (s *Station) onDowngrade(body []byte, major uint16, allowFallback bool) error {
|
||||
gotMajor, gotMinor, err := parseVersion(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if major >= 3 && gotMajor == 2 {
|
||||
if allowFallback && acceptsOther(major, gotMajor, gotMinor) {
|
||||
return errFallback
|
||||
}
|
||||
s.endFatal("Unsupported protocol version "+strconvU(gotMajor)+"."+strconvU(gotMinor), true)
|
||||
|
||||
Reference in New Issue
Block a user