feat(plc4go): name every timeout via utils.WithNamedTimeout

plc4go/internal/ads/Interactions.go

@@ -258,7 +258,7 @@ func (m *Connection) ExecuteAdsDeleteDeviceNotificationRequest(ctx context.Conte
 }
 
 func ReadWithTimeout[T spi.Message](ctx context.Context, ch <-chan T) (T, error) {
-	timeout, cancelFunc := context.WithTimeout(ctx, 5*time.Second)
+	timeout, cancelFunc := utils.WithNamedTimeout(ctx, "read timeout", 5*time.Second)
 	defer cancelFunc()
 
 	select {

plc4go/internal/bacnetip/Discoverer.go

@@ -93,7 +93,7 @@ func (d *Discoverer) Discover(ctx context.Context, callback func(event apiModel.
 	if timeout <= 0 {
 		timeout = 5 * time.Second
 	}
-	ctx, cancelFunc := context.WithTimeout(ctx, timeout)
+	ctx, cancelFunc := utils.WithNamedTimeout(ctx, "discovery timeout", timeout)
 	defer cancelFunc()
 	incomingBVLCChannel, err := d.broadcastAndDiscover(ctx, communicationChannels, specificOptions)
 	if err != nil {

plc4go/internal/bacnetip/ReaderSegmentation.go

@@ -118,7 +118,7 @@ func (m *Reader) expectSegment(ctx context.Context, invokeId uint8) (<-chan read
 	segCh := make(chan readWriteModel.APDUComplexAck, 1)
 	errCh := make(chan error, 1)
 
-	expectCtx, cancel := context.WithTimeout(ctx, segmentWaitTimeout)
+	expectCtx, cancel := utils.WithNamedTimeout(ctx, "segment wait timeout", segmentWaitTimeout)
 
 	m.messageCodec.Expect(expectCtx, "readSegment",
 		func(message spi.Message) bool {

plc4go/internal/bacnetip/SenderSegmentation.go

@@ -28,6 +28,7 @@ import (
 	readWriteModel "github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model"
 	"github.com/apache/plc4x/plc4go/spi"
 	"github.com/apache/plc4x/plc4go/spi/errors"
+	"github.com/apache/plc4x/plc4go/spi/utils"
 )
 
 // segmentAckWaitTimeout bounds how long we wait for the peer's SegmentAck after
@@ -216,7 +217,7 @@ func (s *segmentedRequestSender) expectSegmentAck(ctx context.Context, invokeId
 	ackCh := make(chan readWriteModel.APDUSegmentAck, 1)
 	errCh := make(chan error, 1)
 
-	expectCtx, cancel := context.WithTimeout(ctx, segmentAckWaitTimeout)
+	expectCtx, cancel := utils.WithNamedTimeout(ctx, "segment ack wait timeout", segmentAckWaitTimeout)
 
 	m := s.messageCodec
 	m.Expect(expectCtx, "requestSegmentAck",

plc4go/internal/cbus/Browser.go

@@ -36,6 +36,7 @@ import (
 	"github.com/apache/plc4x/plc4go/spi/errors"
 	spiModel "github.com/apache/plc4x/plc4go/spi/model"
 	"github.com/apache/plc4x/plc4go/spi/options"
+	"github.com/apache/plc4x/plc4go/spi/utils"
 )
 
 type Browser struct {
@@ -122,7 +123,7 @@ unitLoop:
 				AddTag(readTagName, NewCALIdentifyTag(unit, nil /*TODO: add bridge support*/, attribute, 1)).
 				Build()
 			timeout := 5 * time.Second // TODO: do we want to keep this
-			timeoutCtx, timeoutCancel := context.WithTimeout(ctx, timeout)
+			timeoutCtx, timeoutCancel := utils.WithNamedTimeout(ctx, "browse timeout", timeout)
 			m.log.Trace().
 				Stringer("readRequest", readRequest).
 				Dur("timeout", timeout).
@@ -216,7 +217,7 @@ func (m *Browser) getInstalledUnitAddressBytes(ctx context.Context) (map[byte]an
 	if err != nil {
 		return nil, errors.Wrap(err, "Error subscribing to the installation MMI")
 	}
-	subCtx, subCtxCancel := context.WithTimeout(ctx, 2*time.Second)
+	subCtx, subCtxCancel := utils.WithNamedTimeout(ctx, "MMI subscribe timeout", 2*time.Second)
 	defer subCtxCancel()
 	subscriptionResult := <-subscriptionRequest.Execute(subCtx)
 	if err := subscriptionResult.GetErr(); err != nil {
@@ -336,7 +337,7 @@ func (m *Browser) getInstalledUnitAddressBytes(ctx context.Context) (map[byte]an
 	if err != nil {
 		return nil, errors.Wrap(err, "Error building the installation MMI")
 	}
-	readCtx, readCtxCancel := context.WithTimeout(ctx, 2*time.Second)
+	readCtx, readCtxCancel := utils.WithNamedTimeout(ctx, "MMI read timeout", 2*time.Second)
 	defer readCtxCancel()
 	readWg := new(sync.WaitGroup)
 	readWg.Go(func() {

plc4go/internal/cbus/MessageCodec.go

@@ -31,10 +31,11 @@ import (
 
 	readWriteModel "github.com/apache/plc4x/plc4go/protocols/cbus/readwrite/model"
 	"github.com/apache/plc4x/plc4go/spi"
-	"github.com/apache/plc4x/plc4go/spi/default"
+	_default "github.com/apache/plc4x/plc4go/spi/default"
 	"github.com/apache/plc4x/plc4go/spi/errors"
 	"github.com/apache/plc4x/plc4go/spi/options"
 	"github.com/apache/plc4x/plc4go/spi/transports"
+	"github.com/apache/plc4x/plc4go/spi/utils"
 )
 
 //go:generate go tool plc4xGenerator -type=MessageCodec
@@ -148,7 +149,7 @@ func (m *MessageCodec) Receive(ctx context.Context) (spi.Message, error) {
 	confirmation := false
 	// Fill the buffer
 	{
-		fillCtx, fillCtxCancel := context.WithTimeout(ctx, 100*time.Millisecond)
+		fillCtx, fillCtxCancel := utils.WithNamedTimeout(ctx, "buffer fill timeout", 100*time.Millisecond)
 		if err := ti.FillBuffer(fillCtx, func(pos uint, currentByte byte, reader transports.ExtendedReader) (keepGoing bool) {
 			switch currentByte {
 			case

plc4go/internal/eip/Connection.go

@@ -27,11 +27,11 @@ import (
 
 	"github.com/rs/zerolog"
 
-	"github.com/apache/plc4x/plc4go/pkg/api"
+	plc4go "github.com/apache/plc4x/plc4go/pkg/api"
 	apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
 	readWriteModel "github.com/apache/plc4x/plc4go/protocols/eip/readwrite/model"
 	"github.com/apache/plc4x/plc4go/spi"
-	"github.com/apache/plc4x/plc4go/spi/default"
+	_default "github.com/apache/plc4x/plc4go/spi/default"
 	"github.com/apache/plc4x/plc4go/spi/errors"
 	spiModel "github.com/apache/plc4x/plc4go/spi/model"
 	"github.com/apache/plc4x/plc4go/spi/options"
@@ -159,7 +159,7 @@ func (c *Connection) Connect(ctx context.Context) error {
 
 func (c *Connection) Close() error {
 	ctx := context.TODO()
-	ctx, cancelFunc := context.WithTimeout(ctx, 5*time.Second)
+	ctx, cancelFunc := utils.WithNamedTimeout(ctx, "connection close timeout", 5*time.Second)
 	defer cancelFunc()
 	c.log.Debug().Msg("Sending UnregisterSession EIP Packet")
 	if err := c.messageCodec.SendRequest(ctx, "close_eip_disconnect_request", readWriteModel.NewEipDisconnectRequest(c.sessionHandle, 0, []byte(DefaultSenderContext), 0), func(message spi.Message) bool {

plc4go/internal/knxnetip/Connection.go

@@ -43,6 +43,7 @@ import (
 	"github.com/apache/plc4x/plc4go/spi/options"
 	"github.com/apache/plc4x/plc4go/spi/tracer"
 	"github.com/apache/plc4x/plc4go/spi/transports"
+	"github.com/apache/plc4x/plc4go/spi/utils"
 )
 
 //go:generate go tool plc4xGenerator -type=ConnectionMetadata
@@ -381,7 +382,7 @@ func (m *Connection) doSomethingAndClose(something func() error) error {
 
 func (m *Connection) Close() error {
 	ctx := context.TODO()
-	ctx, cancelFunc := context.WithTimeout(ctx, 5*time.Second)
+	ctx, cancelFunc := utils.WithNamedTimeout(ctx, "connection close timeout", 5*time.Second)
 	defer cancelFunc()
 
 	// Stop the connection-state checker.
@@ -411,7 +412,7 @@ func (m *Connection) Close() error {
 
 func (m *Connection) IsConnected() bool {
 	ctx := context.TODO()
-	ctx, cancelFunc := context.WithTimeout(ctx, 5*time.Second)
+	ctx, cancelFunc := utils.WithNamedTimeout(ctx, "connection status check timeout", 5*time.Second)
 	defer cancelFunc()
 
 	if m.messageCodec != nil {

plc4go/internal/opcua/Connection.go

@@ -27,14 +27,15 @@ import (
 
 	"github.com/rs/zerolog"
 
-	"github.com/apache/plc4x/plc4go/pkg/api"
+	plc4go "github.com/apache/plc4x/plc4go/pkg/api"
 	apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
 	"github.com/apache/plc4x/plc4go/spi"
-	"github.com/apache/plc4x/plc4go/spi/default"
+	_default "github.com/apache/plc4x/plc4go/spi/default"
 	"github.com/apache/plc4x/plc4go/spi/errors"
 	spiModel "github.com/apache/plc4x/plc4go/spi/model"
 	"github.com/apache/plc4x/plc4go/spi/options"
 	"github.com/apache/plc4x/plc4go/spi/tracer"
+	"github.com/apache/plc4x/plc4go/spi/utils"
 )
 
 //go:generate go tool plc4xGenerator -type=Connection
@@ -151,7 +152,7 @@ func (c *Connection) Connect(ctx context.Context) error {
 
 func (c *Connection) Close() error {
 	ctx := context.TODO()
-	ctx, cancelFunc := context.WithTimeout(ctx, 5*time.Second)
+	ctx, cancelFunc := utils.WithNamedTimeout(ctx, "connection close timeout", 5*time.Second)
 	defer cancelFunc()
 
 	c.channel.onDisconnect(ctx, c)

plc4go/internal/simulated/Connection.go

@@ -130,7 +130,7 @@ func (c *Connection) Connect(_ context.Context) error {
 
 func (c *Connection) Close() error {
 	ctx := context.TODO()
-	ctx, cancelFunc := context.WithTimeout(ctx, 5*time.Second)
+	ctx, cancelFunc := utils.WithNamedTimeout(ctx, "connection close timeout", 5*time.Second)
 	defer cancelFunc()
 
 	// Check if the connection is connected.

plc4go/pkg/api/cache/PlcConnectionCache.go

@@ -28,12 +28,13 @@ import (
 
 	"github.com/rs/zerolog"
 
-	"github.com/apache/plc4x/plc4go/pkg/api"
+	plc4go "github.com/apache/plc4x/plc4go/pkg/api"
 	"github.com/apache/plc4x/plc4go/pkg/api/config"
 	"github.com/apache/plc4x/plc4go/spi"
 	"github.com/apache/plc4x/plc4go/spi/errors"
 	"github.com/apache/plc4x/plc4go/spi/options"
 	"github.com/apache/plc4x/plc4go/spi/tracer"
+	"github.com/apache/plc4x/plc4go/spi/utils"
 )
 
 type PlcConnectionCache interface {
@@ -288,7 +289,7 @@ func (c *plcConnectionCache) Close() error {
 			// Try to get a lease as this way we kow we're not closing the connection
 			// while some go func is still using it.
 			ccLog.Trace().Msg("getting a lease")
-			ctx, cancel := context.WithTimeout(ctx, c.maxWaitTime)
+			ctx, cancel := utils.WithNamedTimeout(ctx, "lease wait timeout", c.maxWaitTime)
 			connChan, errChan := connectionContainer.lease(ctx)
 			select {
 			// We're just getting the lease as this way we can be sure nobody else is using it.

plc4go/pkg/api/cache/plcConnectionLease.go

@@ -28,6 +28,7 @@ import (
 	apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
 	"github.com/apache/plc4x/plc4go/spi/errors"
 	"github.com/apache/plc4x/plc4go/spi/tracer"
+	"github.com/apache/plc4x/plc4go/spi/utils"
 )
 
 type plcConnectionLease struct {
@@ -78,7 +79,7 @@ func (t *plcConnectionLease) Connect(_ context.Context) error {
 
 func (t *plcConnectionLease) Close() error {
 	ctx := context.TODO()
-	ctx, cancelFunc := context.WithTimeout(ctx, 5*time.Second)
+	ctx, cancelFunc := utils.WithNamedTimeout(ctx, "connection close timeout", 5*time.Second)
 	defer cancelFunc()
 
 	if t.connection == nil {

plc4go/spi/transactions/RequestTransaction.go

@@ -31,6 +31,7 @@ import (
 
 	"github.com/apache/plc4x/plc4go/spi/errors"
 	"github.com/apache/plc4x/plc4go/spi/pool"
+	"github.com/apache/plc4x/plc4go/spi/utils"
 )
 
 // RequestTransaction represents a transaction
@@ -141,7 +142,7 @@ func (t *requestTransaction) Submit(operationInfo string, operation RequestTrans
 
 func (t *requestTransaction) AwaitCompletion(ctx context.Context) error {
 	t.log.Trace().Msg("Awaiting completion")
-	timeout, cancelFunc := context.WithTimeout(ctx, time.Minute*30) // This is intentionally set very high
+	timeout, cancelFunc := utils.WithNamedTimeout(ctx, "transaction completion timeout", time.Minute*30) // This is intentionally set very high
 	defer cancelFunc()
 	for t.getCompletionFuture() == nil {
 		time.Sleep(time.Millisecond * 10)

plc4go/spi/utils/NamedTimeout.go

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package utils
+
+import (
+	"context"
+	"fmt"
+	"time"
+)
+
+// WithNamedTimeout is context.WithTimeout with a legible deadline: when the timer
+// fires, the context's cause reads "<name> <duration> exceeded" instead of the bare
+// context.DeadlineExceeded singleton, so errors say WHICH deadline fired and its
+// value ("segment ack wait timeout 5s exceeded" instead of an anonymous
+// "context deadline exceeded" that could be any of the stacked deadlines).
+//
+// The cause wraps context.DeadlineExceeded. That is load-bearing: consumers such as
+// net/http (Go 1.23+) propagate the cause INSTEAD OF the sentinel, so a cause that
+// does not wrap it would break errors.Is(err, context.DeadlineExceeded) /
+// Timeout() classification downstream. ctx.Err() still returns the plain sentinel,
+// as for every deadline context; the named cause is available via context.Cause.
+func WithNamedTimeout(parent context.Context, name string, d time.Duration) (context.Context, context.CancelFunc) {
+	return context.WithTimeoutCause(parent, d,
+		fmt.Errorf("%s %s exceeded: %w", name, d, context.DeadlineExceeded))
+}

plc4go/spi/utils/NamedTimeout_test.go

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package utils
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestWithNamedTimeout(t *testing.T) {
+	ctx, cancel := WithNamedTimeout(context.Background(), "transaction completion timeout", 10*time.Millisecond)
+	defer cancel()
+	<-ctx.Done()
+
+	// ctx.Err() keeps the plain sentinel so existing callers are unaffected.
+	require.ErrorIs(t, ctx.Err(), context.DeadlineExceeded)
+
+	// The cause names the deadline and its value AND wraps the sentinel so
+	// errors.Is(err, context.DeadlineExceeded) keeps working wherever the cause
+	// replaces the sentinel in an error chain.
+	cause := context.Cause(ctx)
+	assert.ErrorIs(t, cause, context.DeadlineExceeded)
+	assert.Contains(t, cause.Error(), "transaction completion timeout 10ms exceeded")
+}
+
+func TestWithNamedTimeout_cancelBeforeDeadline(t *testing.T) {
+	ctx, cancel := WithNamedTimeout(context.Background(), "connection close timeout", time.Hour)
+	cancel()
+	require.ErrorIs(t, ctx.Err(), context.Canceled)
+	assert.NotContains(t, context.Cause(ctx).Error(), "connection close timeout")
+}