Improve error handling

Author: puzpuzpuz

http_sender.go

@@ -29,7 +29,6 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"math/big"
@@ -176,7 +175,7 @@ func (s *httpLineSender) flush0(ctx context.Context, closing bool) error {
 	)
 
 	if s.closed {
-		return errors.New("cannot flush a closed LineSender")
+		return errClosedSenderFlush
 	}
 
 	err := s.buf.LastErr()
@@ -187,7 +186,7 @@ func (s *httpLineSender) flush0(ctx context.Context, closing bool) error {
 	}
 	if s.buf.HasTable() {
 		s.buf.DiscardPendingMsg()
-		return errors.New("pending ILP message must be finalized with At or AtNow before calling Flush")
+		return errFlushWithPendingMessage
 	}
 
 	if s.buf.msgCount == 0 {
@@ -285,7 +284,7 @@ func (s *httpLineSender) BoolColumn(name string, val bool) LineSender {
 
 func (s *httpLineSender) Close(ctx context.Context) error {
 	if s.closed {
-		return errors.New("double http sender close")
+		return errDoubleSenderClose
 	}
 
 	var err error
@@ -309,7 +308,7 @@ func (s *httpLineSender) AtNow(ctx context.Context) error {
 
 func (s *httpLineSender) At(ctx context.Context, ts time.Time) error {
 	if s.closed {
-		return errors.New("cannot queue new messages on a closed LineSender")
+		return errClosedSenderAt
 	}
 
 	sendTs := true

sender.go

@@ -35,6 +35,13 @@ import (
 	"time"
 )
 
+var (
+	errClosedSenderFlush       = errors.New("cannot flush a closed LineSender")
+	errFlushWithPendingMessage = errors.New("pending ILP message must be finalized with At or AtNow before calling Flush")
+	errClosedSenderAt          = errors.New("cannot queue new messages on a closed LineSender")
+	errDoubleSenderClose       = errors.New("double sender close")
+)
+
 // LineSender allows you to insert rows into QuestDB by sending ILP
 // messages over HTTP or TCP protocol.
 //

sender_pool.go

@@ -35,6 +35,12 @@ import (
 	"time"
 )
 
+var (
+	errAcquireFromClosedPool = errors.New("cannot acquire a LineSender from a closed LineSenderPool")
+	errHttpOnlySender        = errors.New("tcp/s not supported for pooled senders, use http/s only")
+	errPooledSenderClose     = errors.New("error closing one or more LineSenders in the pool")
+)
+
 // LineSenderPool wraps a mutex-protected slice of [LineSender]. It allows a goroutine to
 // Acquire a sender from the pool and Release it back to the pool when it's done being used.
 //
@@ -72,7 +78,7 @@ type LineSenderPoolOption func(*LineSenderPool)
 // [WithMaxSenders] option.
 func PoolFromConf(conf string, opts ...LineSenderPoolOption) (*LineSenderPool, error) {
 	if strings.HasPrefix(conf, "tcp") {
-		return nil, errors.New("tcp/s not supported for pooled senders, use http/s only")
+		return nil, errHttpOnlySender
 	}
 
 	pool := &LineSenderPool{
@@ -145,14 +151,18 @@ func (p *LineSenderPool) Sender(ctx context.Context) (LineSender, error) {
 	defer p.mu.Unlock()
 
 	if p.closed {
-		return nil, errors.New("cannot Acquire a LineSender from a closed LineSenderPool")
+		return nil, errAcquireFromClosedPool
 	}
 
 	// We may have to wait for a free sender
 	for len(p.freeSenders) == 0 && p.numSenders == p.maxSenders {
 		p.cond.Wait()
 	}
 
+	if p.closed {
+		return nil, errAcquireFromClosedPool
+	}
+
 	if len(p.freeSenders) > 0 {
 		// Pop sender off the slice and return it
 		s := p.freeSenders[len(p.freeSenders)-1]
@@ -168,7 +178,7 @@ func (p *LineSenderPool) Sender(ctx context.Context) (LineSender, error) {
 		for _, opt := range p.opts {
 			opt(conf)
 			if conf.senderType == tcpSenderType {
-				return nil, errors.New("tcp/s not supported for pooled senders, use http/s only")
+				return nil, errHttpOnlySender
 			}
 		}
 		s, err = newHttpLineSender(conf)
@@ -247,7 +257,7 @@ func (p *LineSenderPool) Close(ctx context.Context) error {
 		return nil
 	}
 
-	err := errors.New("error closing one or more LineSenders in the pool")
+	err := errPooledSenderClose
 	for _, senderErr := range senderErrors {
 		err = fmt.Errorf("%s %w", err, senderErr)
 	}
@@ -340,7 +350,7 @@ func (ps *pooledSender) Flush(ctx context.Context) error {
 
 func (ps *pooledSender) Close(ctx context.Context) error {
 	if atomic.AddUint64(&ps.tick, 1)&1 == 1 {
-		return errors.New("double pooled sender close")
+		return errDoubleSenderClose
 	}
 	return ps.pool.free(ctx, ps)
 }

tcp_sender.go

@@ -33,7 +33,6 @@ import (
 	"crypto/rand"
 	"crypto/tls"
 	"encoding/base64"
-	"errors"
 	"fmt"
 	"math/big"
 	"net"
@@ -137,7 +136,7 @@ func newTcpLineSender(ctx context.Context, conf *lineSenderConfig) (*tcpLineSend
 
 func (s *tcpLineSender) Close(_ context.Context) error {
 	if s.conn == nil {
-		return errors.New("double tcp sender close")
+		return errDoubleSenderClose
 	}
 	conn := s.conn
 	s.conn = nil
@@ -193,7 +192,7 @@ func (s *tcpLineSender) Flush(ctx context.Context) error {
 	}
 	if s.buf.HasTable() {
 		s.buf.DiscardPendingMsg()
-		return errors.New("pending ILP message must be finalized with At or AtNow before calling Flush")
+		return errFlushWithPendingMessage
 	}
 
 	if err = ctx.Err(); err != nil {