Merge remote-tracking branch 'origin/stable' into v2

Add more architectures for mtg

.gitattributes

@@ -0,0 +1,3 @@
+# git config merge.theirs.name "Always accept theirs"
+# git config merge.theirs.driver "cp %B %A"
+default.pgo binary merge=theirs

.goreleaser.yml

@@ -10,13 +10,15 @@ before:
     - go generate ./...
 
 builds:
-  - binary: '{{ .ProjectName }}'
+  - id: default
+    binary: '{{ .ProjectName }}'
     goos:
       - darwin
       - freebsd
       - linux
       - netbsd
       - openbsd
+      - windows
     goarch:
       - 386
       - amd64
@@ -34,15 +36,92 @@ builds:
     ignore:
       - goos: darwin
         goarch: 386
+      - goos: darwin
+        goarch: arm
       - goos: freebsd
         goarch: arm64
       - goos: netbsd
         goarch: arm64
       - goos: openbsd
         goarch: arm64
+      - goos: windows
+        goarch: 386
+      - goos: windows
+        goarch: arm
+  - id: mips
+    binary: '{{ .ProjectName }}'
+    goos:
+      - linux
+    goarch:
+      - mips
+      - mipsle
+    gomips:
+      - softfloat
+    env:
+      - CGO_ENABLED=0
+    flags:
+      - -trimpath
+      - -mod=readonly
+    ldflags: -s -w -X main.version={{ .Version }}
+  - id: arm64-v9
+    binary: '{{ .ProjectName }}'
+    goos:
+      - darwin
+      - linux
+    goarch:
+      - arm64
+    goarm64:
+      - v9.0
+    env:
+      - CGO_ENABLED=0
+    flags:
+      - -trimpath
+      - -mod=readonly
+    ldflags: -s -w -X main.version={{ .Version }}
+  - id: amd64-v3
+    binary: '{{ .ProjectName }}'
+    goos:
+      - darwin
+      - freebsd
+      - linux
+      - netbsd
+      - openbsd
+      - windows
+    goarch:
+      - amd64
+    goamd64:
+      - v3
+    env:
+      - CGO_ENABLED=0
+    flags:
+      - -trimpath
+      - -mod=readonly
+    ldflags: -s -w -X main.version={{ .Version }}
 
 archives:
-  - name_template: '{{ .ProjectName }}-{{ .Version }}-{{ .Os }}-{{ .Arch }}{{ if .Arm }}v{{ .Arm }}{{ end }}'
+  - id: default
+    ids:
+      - default
+      - mips
+    name_template: '{{ .ProjectName }}-{{ .Version }}-{{ .Os }}-{{ .Arch }}{{ if .Arm }}v{{ .Arm }}{{ end }}'
+    formats:
+      - tar.gz
+    wrap_in_directory: true
+    format_overrides:
+      - goos: windows
+        formats:
+          - zip
+    files:
+      - LICENSE
+      - README.md
+      - SECURITY.md
+      - BEST_PRACTICES.md
+      - example.config.toml
+  - id: optimized
+    ids:
+      - arm64-v9
+      - amd64-v3
+    name_template: '{{ .ProjectName }}-{{ .Version }}-{{ .Os }}-{{ .Arch }}{{ if .Arm64 }}-{{ .Arm64 }}{{ end }}{{ if .Amd64 }}-{{ .Amd64 }}{{ end }}'
     formats:
       - tar.gz
     wrap_in_directory: true

README.md

@@ -380,6 +380,7 @@ ExecStart=/usr/local/bin/mtg run /etc/mtg.toml
 Restart=always
 RestartSec=3
 DynamicUser=true
+LimitNOFILE=65536
 AmbientCapabilities=CAP_NET_BIND_SERVICE
 
 [Install]

main.go

@@ -14,6 +14,23 @@ import (
 )
 
 func main() {
+	// this runs profiling server. To enable it, build with prof tag
+	//   $ go build -tags prof
+	//
+	// Then you can pass a port using MTG_PROF_PORT environment variable.
+	// Default is 6000
+	//   $ MTG_PROF_PORT=6000 mtg run config.toml
+	//
+	// It will run a webserver with profiling data on
+	// localhost:${MTG_PROF_PORT:-6000}.
+	//
+	// To collect PGO do following:
+	//   $ curl -o default.pgo 'http://localhost:6000/debug/pprof/profile?seconds=300'
+	//
+	// See also https://pkg.go.dev/net/http/pprof
+	//          https://go.dev/blog/pprof
+	runProfile()
+
 	cli := &cli.CLI{}
 	ctx := kong.Parse(cli, kong.Vars{
 		"version": getVersion(),

mtglib/internal/doppel/conn.go

@@ -16,48 +16,25 @@ type Conn struct {
 }
 
 type connPayload struct {
-	ctx           context.Context
+	ctx         context.Context
-	ctxCancel     context.CancelCauseFunc
+	ctxCancel   context.CancelCauseFunc
-	clock         Clock
+	clock       Clock
-	wg            sync.WaitGroup
+	wg          sync.WaitGroup
-	syncWriteLock sync.RWMutex
+	writeStream bytes.Buffer
-	writeStream   bytes.Buffer
+	writtenCond sync.Cond
-	writeCond     *sync.Cond
+	done        bool
 }
 
 func (c Conn) Write(p []byte) (int, error) {
-	c.p.syncWriteLock.RLock()
+	if len(p) == 0 {
-	defer c.p.syncWriteLock.RUnlock()
+		return 0, context.Cause(c.p.ctx)
+	}
 
-	c.p.writeCond.L.Lock()
+	c.p.writtenCond.L.Lock()
 	c.p.writeStream.Write(p)
-	c.p.writeCond.L.Unlock()
+	c.p.writtenCond.L.Unlock()
 
-	return len(p), context.Cause(c.p.ctx)
+	c.p.writtenCond.Signal()
-}
-
-func (c Conn) SyncWrite(p []byte) (int, error) {
-	c.p.syncWriteLock.Lock()
-	defer c.p.syncWriteLock.Unlock()
-
-	c.p.writeCond.L.Lock()
-	// wait until buffer is exhausted
-	for c.p.writeStream.Len() != 0 && context.Cause(c.p.ctx) == nil {
-		c.p.writeCond.Wait()
-	}
-	c.p.writeStream.Write(p)
-	c.p.writeCond.L.Unlock()
-
-	if err := context.Cause(c.p.ctx); err != nil {
-		return len(p), err
-	}
-
-	c.p.writeCond.L.Lock()
-	// wait until data will be sent
-	for c.p.writeStream.Len() != 0 && context.Cause(c.p.ctx) == nil {
-		c.p.writeCond.Wait()
-	}
-	c.p.writeCond.L.Unlock()
 
 	return len(p), context.Cause(c.p.ctx)
 }
@@ -69,8 +46,6 @@ func (c Conn) Start() {
 }
 
 func (c Conn) start() {
-	defer c.p.writeCond.Broadcast()
-
 	buf := [tls.MaxRecordSize]byte{}
 
 	for {
@@ -80,25 +55,34 @@ func (c Conn) start() {
 		case <-c.p.clock.tick:
 		}
 
-		c.p.writeCond.L.Lock()
+		size := c.p.clock.stats.Size()
-		n, err := c.p.writeStream.Read(buf[:c.p.clock.stats.Size()])
-		c.p.writeCond.L.Unlock()
 
-		if n == 0 || err != nil {
+		c.p.writtenCond.L.Lock()
+		for c.p.writeStream.Len() == 0 && !c.p.done {
+			c.p.writtenCond.Wait()
+		}
+		n, _ := c.p.writeStream.Read(buf[tls.SizeHeader : tls.SizeHeader+size])
+		c.p.writtenCond.L.Unlock()
+
+		if n == 0 {
 			continue
 		}
 
-		if err := tls.WriteRecord(c.Conn, buf[:n]); err != nil {
+		if err := tls.WriteRecordInPlace(c.Conn, buf[:], n); err != nil {
 			c.p.ctxCancel(err)
 			return
 		}
-
-		c.p.writeCond.Signal()
 	}
 }
 
 func (c Conn) Stop() {
 	c.p.ctxCancel(nil)
+
+	c.p.writtenCond.L.Lock()
+	c.p.done = true
+	c.p.writtenCond.L.Unlock()
+	c.p.writtenCond.Broadcast()
+
 	c.p.wg.Wait()
 }
 
@@ -109,7 +93,9 @@ func NewConn(ctx context.Context, conn essentials.Conn, stats *Stats) Conn {
 		p: &connPayload{
 			ctx:       ctx,
 			ctxCancel: cancel,
-			writeCond: sync.NewCond(&sync.Mutex{}),
+			writtenCond: sync.Cond{
+				L: &sync.Mutex{},
+			},
 			clock: Clock{
 				stats: stats,
 				tick:  make(chan struct{}),

mtglib/internal/doppel/conn_test.go

@@ -141,6 +141,37 @@ func (suite *ConnTestSuite) TestWriteReturnsErrorAfterStop() {
 	suite.Error(err)
 }
 
+func (suite *ConnTestSuite) TestStopDoesNotDeadlockWhenStartIsWaiting() {
+	suite.connMock.
+		On("Write", mock.AnythingOfType("[]uint8")).
+		Return(0, nil).
+		Maybe()
+
+	for range 100 {
+		func() {
+			ctx, cancel := context.WithCancel(suite.ctx)
+			defer cancel()
+
+			c := NewConn(ctx, suite.connMock, &Stats{
+				k:      2.0,
+				lambda: 0.01,
+			})
+
+			done := make(chan struct{})
+			go func() {
+				defer close(done)
+				c.Stop()
+			}()
+
+			select {
+			case <-done:
+			case <-time.After(2 * time.Second):
+				suite.Fail("Stop() deadlocked: start() likely stuck in writtenCond.Wait()")
+			}
+		}()
+	}
+}
+
 func (suite *ConnTestSuite) TestStopOnUnderlyingWriteError() {
 	suite.connMock.
 		On("Write", mock.AnythingOfType("[]uint8")).
@@ -157,136 +188,6 @@ func (suite *ConnTestSuite) TestStopOnUnderlyingWriteError() {
 	}, 2*time.Second, time.Millisecond)
 }
 
-func (suite *ConnTestSuite) TestSyncWriteDataSent() {
-	suite.connMock.
-		On("Write", mock.AnythingOfType("[]uint8")).
-		Return(0, nil).
-		Maybe()
-
-	c := suite.makeConn()
-	defer c.Stop()
-
-	payload := []byte("sync hello")
-	n, err := c.SyncWrite(payload)
-	suite.NoError(err)
-	suite.Equal(len(payload), n)
-
-	// SyncWrite returns only after data is flushed to the wire.
-	assembled := &bytes.Buffer{}
-	reader := bytes.NewReader(suite.connMock.Written())
-
-	for {
-		header := make([]byte, tls.SizeHeader)
-		if _, err := io.ReadFull(reader, header); err != nil {
-			break
-		}
-
-		suite.Equal(byte(tls.TypeApplicationData), header[0])
-
-		length := binary.BigEndian.Uint16(header[tls.SizeRecordType+tls.SizeVersion:])
-		rec := make([]byte, length)
-		_, err := io.ReadFull(reader, rec)
-		suite.NoError(err)
-
-		assembled.Write(rec)
-	}
-
-	suite.Equal(payload, assembled.Bytes())
-}
-
-func (suite *ConnTestSuite) TestSyncWriteDrainsBufferFirst() {
-	suite.connMock.
-		On("Write", mock.AnythingOfType("[]uint8")).
-		Return(0, nil).
-		Maybe()
-
-	c := suite.makeConn()
-	defer c.Stop()
-
-	// Buffer some data via async Write.
-	_, err := c.Write([]byte("first"))
-	suite.NoError(err)
-
-	// SyncWrite must drain "first" before sending "second".
-	n, err := c.SyncWrite([]byte("second"))
-	suite.NoError(err)
-	suite.Equal(6, n)
-
-	// All data should be on the wire now.
-	assembled := &bytes.Buffer{}
-	reader := bytes.NewReader(suite.connMock.Written())
-
-	for {
-		header := make([]byte, tls.SizeHeader)
-		if _, err := io.ReadFull(reader, header); err != nil {
-			break
-		}
-
-		length := binary.BigEndian.Uint16(header[tls.SizeRecordType+tls.SizeVersion:])
-		rec := make([]byte, length)
-		_, err := io.ReadFull(reader, rec)
-		suite.NoError(err)
-
-		assembled.Write(rec)
-	}
-
-	suite.Equal([]byte("firstsecond"), assembled.Bytes())
-}
-
-func (suite *ConnTestSuite) TestSyncWriteBlocksAsyncWrite() {
-	suite.connMock.
-		On("Write", mock.AnythingOfType("[]uint8")).
-		Return(0, nil).
-		Maybe()
-
-	c := suite.makeConn()
-	defer c.Stop()
-
-	// Start SyncWrite — it holds exclusive lock.
-	syncDone := make(chan struct{})
-
-	go func() {
-		defer close(syncDone)
-		c.SyncWrite([]byte("exclusive")) //nolint: errcheck
-	}()
-
-	// Give SyncWrite time to acquire the lock.
-	time.Sleep(10 * time.Millisecond)
-
-	// Async Write should block until SyncWrite completes.
-	writeDone := make(chan struct{})
-
-	go func() {
-		defer close(writeDone)
-		c.Write([]byte("blocked")) //nolint: errcheck
-	}()
-
-	// SyncWrite should finish first.
-	<-syncDone
-
-	select {
-	case <-writeDone:
-		// Write completed after SyncWrite — correct.
-	case <-time.After(2 * time.Second):
-		suite.Fail("async Write did not unblock after SyncWrite completed")
-	}
-}
-
-func (suite *ConnTestSuite) TestSyncWriteReturnsErrorAfterStop() {
-	suite.connMock.
-		On("Write", mock.AnythingOfType("[]uint8")).
-		Return(0, nil).
-		Maybe()
-
-	c := suite.makeConn()
-	c.Stop()
-
-	time.Sleep(10 * time.Millisecond)
-
-	_, err := c.SyncWrite([]byte("too late"))
-	suite.Error(err)
-}
-
 func TestConn(t *testing.T) {
 	t.Parallel()
 	suite.Run(t, &ConnTestSuite{})

mtglib/internal/doppel/ganger.go

@@ -98,7 +98,8 @@ func (g *Ganger) run() {
 			g.durations = append(g.durations, durations...)
 
 			if len(g.durations) > DoppelGangerMaxDurations {
-				g.durations = g.durations[len(g.durations)-DoppelGangerMaxDurations:]
+				copy(g.durations, g.durations[len(g.durations)-DoppelGangerMaxDurations:])
+				g.durations = g.durations[:DoppelGangerMaxDurations]
 			}
 
 			if len(g.durations) < MinDurationsToCalculate {

mtglib/internal/tls/conn.go

@@ -34,7 +34,6 @@ type Conn struct {
 
 type connPayload struct {
 	readBuf      bytes.Buffer
-	writeBuf     bytes.Buffer
 	connBuffered *bufio.Reader
 	read         bool
 	write        bool
@@ -80,7 +79,6 @@ func New(conn essentials.Conn, read, write bool) Conn {
 	}
 
 	newConn.p.readBuf.Grow(DefaultBufferSize)
-	newConn.p.writeBuf.Grow(DefaultBufferSize)
 
 	return newConn
 }

mtglib/internal/tls/utils.go

@@ -29,20 +29,24 @@ func ReadRecord(r io.Reader, w io.Writer) (byte, int64, error) {
 
 func WriteRecord(w io.Writer, payload []byte) error {
 	buf := [MaxRecordSize]byte{}
-	buf[0] = TypeApplicationData
+	copy(buf[SizeHeader:], payload)
 
-	bufV := buf[SizeRecordType:]
+	return WriteRecordInPlace(w, buf[:], len(payload))
-	copy(bufV[:SizeVersion], TLSVersion[:])
+}
 
-	bufS := bufV[SizeVersion:]
+func WriteRecordInPlace(w io.Writer, buf []byte, payloadLen int) error {
-	binary.BigEndian.PutUint16(bufS[:SizeSize], uint16(len(payload)))
+	if payloadLen > MaxRecordPayloadSize {
-
+		return fmt.Errorf("payload %d exceeds max %d", payloadLen, MaxRecordPayloadSize)
-	bufP := buf[SizeHeader:]
-	if n := copy(bufP, payload); n != len(payload) {
-		return fmt.Errorf("copied %d bytes of payload instead of %d", n, len(payload))
 	}
 
-	_, err := w.Write(buf[:SizeHeader+len(payload)])
+	buf[0] = TypeApplicationData
+	copy(buf[SizeRecordType:SizeRecordType+SizeVersion], TLSVersion[:])
+	binary.BigEndian.PutUint16(
+		buf[SizeRecordType+SizeVersion:SizeRecordType+SizeVersion+SizeSize],
+		uint16(payloadLen),
+	)
+
+	_, err := w.Write(buf[:SizeHeader+payloadLen])
 
 	return err
 }

mtglib/internal/tls/utils_test.go

@@ -119,6 +119,84 @@ func (suite *UtilsTestSuite) TestWriteRecordPayloadTooLarge() {
 	suite.Error(err)
 }
 
+func (suite *UtilsTestSuite) TestWriteRecordInPlace() {
+	payload := []byte("hello in-place")
+
+	var buf [MaxRecordSize]byte
+	copy(buf[SizeHeader:], payload)
+
+	err := WriteRecordInPlace(suite.dst, buf[:], len(payload))
+	suite.NoError(err)
+
+	written := suite.dst.Bytes()
+	suite.Equal(byte(TypeApplicationData), written[0])
+	suite.Equal(TLSVersion[:], written[SizeRecordType:SizeRecordType+SizeVersion])
+
+	length := binary.BigEndian.Uint16(written[SizeRecordType+SizeVersion:])
+	suite.Equal(uint16(len(payload)), length)
+	suite.Equal(payload, written[SizeHeader:])
+}
+
+func (suite *UtilsTestSuite) TestWriteRecordInPlaceRoundTrip() {
+	payload := []byte("round trip in-place")
+
+	var buf [MaxRecordSize]byte
+	copy(buf[SizeHeader:], payload)
+
+	var wire bytes.Buffer
+
+	err := WriteRecordInPlace(&wire, buf[:], len(payload))
+	suite.NoError(err)
+
+	var recovered bytes.Buffer
+
+	recordType, length, err := ReadRecord(&wire, &recovered)
+	suite.NoError(err)
+	suite.Equal(byte(TypeApplicationData), recordType)
+	suite.Equal(int64(len(payload)), length)
+	suite.Equal(payload, recovered.Bytes())
+}
+
+func (suite *UtilsTestSuite) TestWriteRecordInPlacePayloadTooLarge() {
+	var buf [MaxRecordSize]byte
+
+	err := WriteRecordInPlace(suite.dst, buf[:], MaxRecordPayloadSize+1)
+	suite.Error(err)
+}
+
+func (suite *UtilsTestSuite) TestWriteRecordInPlacePropagatesError() {
+	m := &WriterMock{}
+	m.
+		On("Write", mock.AnythingOfType("[]uint8")).
+		Once().
+		Return(0, errors.New("disk full"))
+
+	var buf [MaxRecordSize]byte
+	copy(buf[SizeHeader:], []byte("data"))
+
+	err := WriteRecordInPlace(m, buf[:], 4)
+	suite.Error(err)
+
+	m.AssertExpectations(suite.T())
+}
+
+func (suite *UtilsTestSuite) TestWriteRecordInPlaceMatchesWriteRecord() {
+	payload := []byte("equivalence check")
+
+	var legacy bytes.Buffer
+	err := WriteRecord(&legacy, payload)
+	suite.NoError(err)
+
+	var buf [MaxRecordSize]byte
+	copy(buf[SizeHeader:], payload)
+
+	var inPlace bytes.Buffer
+	err = WriteRecordInPlace(&inPlace, buf[:], len(payload))
+	suite.NoError(err)
+
+	suite.Equal(legacy.Bytes(), inPlace.Bytes())
+}
+
 func TestUtils(t *testing.T) {
 	t.Parallel()
 	suite.Run(t, &UtilsTestSuite{})

mtglib/proxy.go

@@ -259,9 +259,16 @@ func (p *Proxy) doTelegramCall(ctx *streamContext) error {
 		ctx:      ctx,
 	}
 
+	telegramHost, _, err := net.SplitHostPort(foundAddr.Address)
+	if err != nil {
+		conn.Close() //nolint: errcheck
+
+		return fmt.Errorf("cannot parse telegram address %s: %w", foundAddr.Address, err)
+	}
+
 	p.eventStream.Send(ctx,
 		NewEventConnectedToDC(ctx.streamID,
-			conn.RemoteAddr().(*net.TCPAddr).IP, //nolint: forcetypeassert
+			net.ParseIP(telegramHost),
 			ctx.dc),
 	)
 

run_profile.go

@@ -0,0 +1,7 @@
+//go:build !prof
+
+package main
+
+func runProfile() {
+
+}

run_profile_tag_prof.go

@@ -0,0 +1,27 @@
+//go:build prof
+
+package main
+
+import (
+	"fmt"
+	"net"
+	"net/http"
+	_ "net/http/pprof" //nolint: gosec
+	"os"
+)
+
+const DefaultProfPort = "6000"
+
+func runProfile() {
+	port := os.Getenv("MTG_PROF_PORT")
+	if port == "" {
+		port = DefaultProfPort
+	}
+
+	listener, err := net.Listen("tcp", net.JoinHostPort("127.0.0.1", port))
+	if err != nil {
+		panic(err)
+	}
+
+	go http.Serve(listener, nil)
+}