Merge branch 'master' into dev

Author: foxcpp

.github/workflows/release.yml

@@ -144,7 +144,7 @@ jobs:
         id: docker
         with:
           context: .
-          platforms: linux/amd64 #,linux/arm64  Temporary disabled due to SIGSEGV in gcc.
+          platforms: linux/amd64,linux/arm64
           file: Dockerfile
           push: true
           tags: ${{ steps.meta.outputs.tags }}

.version

@@ -1 +1 @@
-0.9.4
+0.9.5

framework/log/log.go

@@ -205,6 +205,15 @@ func (l *Logger) Write(s []byte) (int, error) {
 	return len(s), nil
 }
 
+// Close closes underlying output in Out.
+func (l *Logger) Close() error {
+	if l.Out == nil {
+		return nil
+	}
+
+	return l.Out.Close()
+}
+
 // DebugWriter returns a writer that will act like Logger.Write
 // but will use debug flag on messages. If Logger.Debug is false,
 // Write method of returned object will be no-op.

internal/endpoint/smtp/smtp.go

@@ -274,7 +274,7 @@ func (endp *Endpoint) setConfig(cfg *config.Map) error {
 	cfg.Bool("defer_sender_reject", false, true, &endp.deferServerReject)
 	cfg.Int("max_logged_rcpt_errors", false, false, 5, &endp.maxLoggedRcptErrors)
 	cfg.Custom("limits", false, false, func() (interface{}, error) {
-		return &limits.Group{}, nil
+		return limits.Empty(endp.log.Sublogger("limits")), nil
 	}, func(cfg *config.Map, n config.Node) (interface{}, error) {
 		var g *limits.Group
 		if err := modconfig.GroupFromNode("limits", n.Args, n, cfg.Globals, &g); err != nil {

internal/limits/limits.go

@@ -28,18 +28,21 @@ package limits
 
 import (
 	"context"
+	"fmt"
 	"net"
 	"strconv"
 	"time"
 
 	"github.com/foxcpp/maddy/framework/config"
 	"github.com/foxcpp/maddy/framework/container"
+	"github.com/foxcpp/maddy/framework/log"
 	"github.com/foxcpp/maddy/framework/module"
 	"github.com/foxcpp/maddy/framework/module/modules"
 	"github.com/foxcpp/maddy/internal/limits/limiters"
 )
 
 type Group struct {
+	log      *log.Logger
 	instName string
 
 	global limiters.MultiLimit
@@ -48,8 +51,15 @@ type Group struct {
 	dest   *limiters.BucketSet // BucketSet of MultiLimit
 }
 
+func Empty(log *log.Logger) *Group {
+	return &Group{
+		log: log,
+	}
+}
+
 func New(c *container.C, _, instName string) (module.Module, error) {
 	return &Group{
+		log:      c.DefaultLogger.Sublogger("limits"),
 		instName: instName,
 	}, nil
 }
@@ -177,22 +187,28 @@ func (g *Group) TakeMsg(ctx context.Context, addr net.IP, sourceDomain string) e
 	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
 
+	g.log.DebugMsg("global TakeContext")
 	if err := g.global.TakeContext(ctx); err != nil {
-		return err
+		return fmt.Errorf("TakeMsg: global: %w", err)
 	}
+	g.log.DebugMsg("global TakeContext done")
 
 	if g.ip != nil {
+		g.log.DebugMsg("ip TakeContext", "ip", addr.String())
 		if err := g.ip.TakeContext(ctx, addr.String()); err != nil {
 			g.global.Release()
-			return err
+			return fmt.Errorf("TakeMsg: ip: %w", err)
 		}
+		g.log.DebugMsg("ip TakeContext done", "ip", addr.String())
 	}
 	if g.source != nil {
+		g.log.DebugMsg("source TakeContext", "domain", sourceDomain)
 		if err := g.source.TakeContext(ctx, sourceDomain); err != nil {
 			g.global.Release()
 			g.ip.Release(addr.String())
-			return err
+			return fmt.Errorf("TakeMSg: source: %w", err)
 		}
+		g.log.DebugMsg("source TakeContext done", "domain", sourceDomain)
 	}
 	return nil
 }
@@ -201,17 +217,27 @@ func (g *Group) TakeDest(ctx context.Context, domain string) error {
 	if g.dest == nil {
 		return nil
 	}
+
 	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
 	defer cancel()
-	return g.dest.TakeContext(ctx, domain)
+
+	g.log.DebugMsg("TakeDest", "domain", domain)
+	if err := g.dest.TakeContext(ctx, domain); err != nil {
+		return fmt.Errorf("TakeDest: dest: %w", err)
+	}
+	g.log.DebugMsg("TakeDest done", "domain", domain)
+	return nil
 }
 
 func (g *Group) ReleaseMsg(addr net.IP, sourceDomain string) {
+	g.log.DebugMsg("global ReleaseMsg")
 	g.global.Release()
 	if g.ip != nil {
+		g.log.DebugMsg("ip ReleaseMsg", "ip", addr.String())
 		g.ip.Release(addr.String())
 	}
 	if g.source != nil {
+		g.log.DebugMsg("source ReleaseMsg", "domain", sourceDomain)
 		g.source.Release(sourceDomain)
 	}
 }
@@ -220,7 +246,10 @@ func (g *Group) ReleaseDest(domain string) {
 	if g.dest == nil {
 		return
 	}
+
+	g.log.DebugMsg("ReleaseDest", "domain", domain)
 	g.dest.Release(domain)
+	g.log.DebugMsg("ReleaseDest done", "domain", domain)
 }
 
 func (g *Group) Name() string {

internal/msgpipeline/msgpipeline.go

@@ -90,6 +90,7 @@ func New(globals map[string]interface{}, cfg []config.Node) (*MsgPipeline, error
 	return &MsgPipeline{
 		msgpipelineCfg: parsedCfg,
 		Resolver:       dns.DefaultResolver(),
+		Log:            log.DefaultLogger.Sublogger("msgpipeline"),
 	}, err
 }
 

internal/smtpconn/smtpconn.go

@@ -239,7 +239,7 @@ func (c *C) attemptConnect(ctx context.Context, lmtp bool, endp config.Endpoint,
 	conn, err = c.Dialer(dialCtx, endp.Network(), endp.Address())
 	cancel()
 	if err != nil {
-		return false, nil, nil, err
+		return false, nil, nil, fmt.Errorf("dialer: %w", err)
 	}
 
 	if endp.IsTLS() {

internal/target/remote/connect.go

@@ -45,6 +45,7 @@ type mxConn struct {
 	errored bool
 
 	reuseLimit int
+	takeDest   bool
 
 	// Amount of times connection was used for an SMTP transaction.
 	transactions int
@@ -207,6 +208,10 @@ func (rd *remoteDelivery) attemptMX(ctx context.Context, conn *mxConn, record *n
 }
 
 func (rd *remoteDelivery) closeConn(c *mxConn) {
+	if c.takeDest {
+		rd.rt.limits.ReleaseDest(c.domain)
+	}
+
 	if err := c.Close(); err != nil {
 		rd.log.Error("client connection close failed", err)
 	}
@@ -270,6 +275,7 @@ func (rd *remoteDelivery) connectionForDomain(ctx context.Context, domain string
 		return nil, err
 	}
 	region.End()
+	conn.takeDest = true
 
 	// Relaxed REQUIRETLS mode is not conforming to the specification strictly
 	// but allows to start deploying client support for REQUIRETLS without the

internal/target/remote/remote.go

@@ -130,7 +130,7 @@ func (rt *Target) Configure(inlineArgs []string, cfg *config.Map) error {
 		return p.L, nil
 	}, &rt.policies)
 	cfg.Custom("limits", false, false, func() (interface{}, error) {
-		return &limits.Group{}, nil
+		return limits.Empty(rt.log.Sublogger("limits")), nil
 	}, func(cfg *config.Map, n config.Node) (interface{}, error) {
 		var g *limits.Group
 		if err := modconfig.GroupFromNode("limits", n.Args, n, cfg.Globals, &g); err != nil {
@@ -455,6 +455,7 @@ func (rd *remoteDelivery) Commit(ctx context.Context) error {
 func (rd *remoteDelivery) Close() error {
 	for _, conn := range rd.connections {
 		rd.rt.limits.ReleaseDest(conn.domain)
+		conn.takeDest = false
 		conn.transactions++
 
 		if !conn.Usable() {

internal/target/remote/remote_test.go

@@ -62,7 +62,7 @@ func testTarget(t *testing.T, zones map[string]mockdns.Zone, extResolver *dns.Ex
 		tlsConfig:   &tls.Config{},
 		log:         testutils.Logger(t, "remote"),
 		policies:    extraPolicies,
-		limits:      &limits.Group{},
+		limits:      limits.Empty(testutils.Logger(t, "limits")),
 		pool: pool.New(pool.Config{
 			MaxKeys:             5000,
 			MaxConnsPerKey:      5,      // basically, max. amount of idle connections in cache