diff --git a/plc4go/internal/ads/Interactions.go b/plc4go/internal/ads/Interactions.go index acfde00629..c308dffbd5 100644 --- a/plc4go/internal/ads/Interactions.go +++ b/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 { diff --git a/plc4go/internal/bacnetip/Discoverer.go b/plc4go/internal/bacnetip/Discoverer.go index e62c829fbe..8f3d2a31ba 100644 --- a/plc4go/internal/bacnetip/Discoverer.go +++ b/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 { diff --git a/plc4go/internal/bacnetip/ReaderSegmentation.go b/plc4go/internal/bacnetip/ReaderSegmentation.go index c3a2ee375c..536edc0ef2 100644 --- a/plc4go/internal/bacnetip/ReaderSegmentation.go +++ b/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 { diff --git a/plc4go/internal/bacnetip/SenderSegmentation.go b/plc4go/internal/bacnetip/SenderSegmentation.go index 85b7efd8fe..73d93a2a08 100644 --- a/plc4go/internal/bacnetip/SenderSegmentation.go +++ b/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", diff --git a/plc4go/internal/cbus/Browser.go b/plc4go/internal/cbus/Browser.go index f024cc82bc..d9d1ed6980 100644 --- a/plc4go/internal/cbus/Browser.go +++ b/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() { diff --git a/plc4go/internal/cbus/MessageCodec.go b/plc4go/internal/cbus/MessageCodec.go index 2dcd8fba9b..3cd3e1de1f 100644 --- a/plc4go/internal/cbus/MessageCodec.go +++ b/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 diff --git a/plc4go/internal/eip/Connection.go b/plc4go/internal/eip/Connection.go index 3c1d6ed396..862d511365 100644 --- a/plc4go/internal/eip/Connection.go +++ b/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 { diff --git a/plc4go/internal/knxnetip/Connection.go b/plc4go/internal/knxnetip/Connection.go index a71470b70d..edae39b0d0 100644 --- a/plc4go/internal/knxnetip/Connection.go +++ b/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 { diff --git a/plc4go/internal/opcua/Connection.go b/plc4go/internal/opcua/Connection.go index e1e43100b5..e5fd4e0dfb 100644 --- a/plc4go/internal/opcua/Connection.go +++ b/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) diff --git a/plc4go/internal/simulated/Connection.go b/plc4go/internal/simulated/Connection.go index b18a9e0c4b..da97489557 100644 --- a/plc4go/internal/simulated/Connection.go +++ b/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. diff --git a/plc4go/pkg/api/cache/PlcConnectionCache.go b/plc4go/pkg/api/cache/PlcConnectionCache.go index 3f19b54b2f..64690de16f 100644 --- a/plc4go/pkg/api/cache/PlcConnectionCache.go +++ b/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. diff --git a/plc4go/pkg/api/cache/plcConnectionLease.go b/plc4go/pkg/api/cache/plcConnectionLease.go index 8094ec31b8..5f23b3c4e2 100644 --- a/plc4go/pkg/api/cache/plcConnectionLease.go +++ b/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 { diff --git a/plc4go/spi/transactions/RequestTransaction.go b/plc4go/spi/transactions/RequestTransaction.go index c52ace0e17..89f486e008 100644 --- a/plc4go/spi/transactions/RequestTransaction.go +++ b/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) diff --git a/plc4go/spi/utils/NamedTimeout.go b/plc4go/spi/utils/NamedTimeout.go new file mode 100644 index 0000000000..a5cb7d164d --- /dev/null +++ b/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 " 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)) +} diff --git a/plc4go/spi/utils/NamedTimeout_test.go b/plc4go/spi/utils/NamedTimeout_test.go new file mode 100644 index 0000000000..642ef94424 --- /dev/null +++ b/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") +}