Repository: v2fly/v2ray-core
File: proxy/shadowsocks2022/client_session.go (Go)

Code before:
	if err != nil {
		return 0, newError("unable to write to parent session").Base(err)
	}
	return len(p), nil
}

func (c *ClientUDPSessionConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
	for {
		select {
		case <-c.ctx.Done():
			return 0, nil, io.EOF
		case resp := <-c.readChan:
			n = copy(p, resp.Payload.Bytes())
			resp.Payload.Release()

			var trackedState *ClientUDPSessionServerTracker
			if trackedStateReceived, ok := c.trackedServerSessionID[string(resp.SessionID[:])]; !ok {
				expiredServerSessionID := make([]string, 0)
				for key, value := range c.trackedServerSessionID {
					if time.Since(value.lastSeen) > 125*time.Second {
						expiredServerSessionID = append(expiredServerSessionID, key)
					}
				}
				for _, key := range expiredServerSessionID {
					delete(c.trackedServerSessionID, key)
				}

				state := &ClientUDPSessionServerTracker{
					rxReplayDetector: replaydetector.New(1024, ^uint64(0)),
				}
				c.trackedServerSessionID[string(resp.SessionID[:])] = state
				c.parent.locker.RLock()
				c.parent.sessionMapAlias[string(resp.SessionID[:])] = string(resp.ClientSessionID[:])
				c.parent.locker.RUnlock()
				trackedState = state
			} else {
				trackedState = trackedStateReceived
			}

			if accept, ok := trackedState.rxReplayDetector.Check(resp.PacketID); ok {
				accept()
			} else {
				newError("misbehaving server: replayed packet").Base(err).WriteToLog()
				continue
			}
			trackedState.lastSeen = time.Now()

			addr = &net.UDPAddr{IP: resp.Address.IP(), Port: resp.Port}
		}
		return n, addr, nil
	}

Review comment (performance):
It's fine to delete k-v pairs from a map while iterating over it. There's no need to cache the keys to delete in a separate slice.

The replay window during a server session change is only 60 seconds. Why does it cache the state for 125 seconds?

Code after:
	if err != nil {
		return 0, newError("unable to write to parent session").Base(err)
	}
	return len(p), nil
}

func (c *ClientUDPSessionConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
	for {
		select {
		case <-c.ctx.Done():
			return 0, nil, io.EOF
		case resp := <-c.readChan:
			n = copy(p, resp.Payload.Bytes())
			resp.Payload.Release()

			var trackedState *ClientUDPSessionServerTracker
			if trackedStateReceived, ok := c.trackedServerSessionID[string(resp.SessionID[:])]; !ok {
				for key, value := range c.trackedServerSessionID {
					if time.Since(value.lastSeen) > 65*time.Second {
						delete(c.trackedServerSessionID, key)
					}
				}

				state := &ClientUDPSessionServerTracker{
					rxReplayDetector: replaydetector.New(1024, ^uint64(0)),
				}
				c.trackedServerSessionID[string(resp.SessionID[:])] = state
				c.parent.locker.RLock()
				c.parent.sessionMapAlias[string(resp.SessionID[:])] = string(resp.ClientSessionID[:])
				c.parent.locker.RUnlock()
				trackedState = state
			} else {
				trackedState = trackedStateReceived
			}

			if accept, ok := trackedState.rxReplayDetector.Check(resp.PacketID); ok {
				accept()
			} else {
				newError("misbehaving server: replayed packet").Base(err).WriteToLog()
				continue
			}
			trackedState.lastSeen = time.Now()

			addr = &net.UDPAddr{IP: resp.Address.IP(), Port: resp.Port}
		}
		return n, addr, nil
	}
}