This is an automated email from the ASF dual-hosted git repository. sruehl pushed a commit to branch feat/timeout_naming in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit bb3557b0c4dbfaf7a2fec026525654ee7c332a33 Author: Sebastian Rühl <[email protected]> AuthorDate: Wed Aug 12 09:40:30 2026 +0200 feat(plc4go): name every timeout via utils.WithNamedTimeout context.DeadlineExceeded is a metadata-free singleton: a bare "context deadline exceeded" never says which of the stacked deadlines fired or what its value was, making field timeouts undiagnosable. utils.WithNamedTimeout wraps context.WithTimeoutCause with a cause that reads "<name> <duration> exceeded" AND wraps the sentinel, so context.Cause carries the diagnosis while errors.Is(err, context.DeadlineExceeded) keeps working for every consumer (net/http propagates the cause in place of the sentinel since Go 1.23). All 16 bare context.WithTimeout call sites are converted with descriptive names (transaction completion timeout, segment ack wait timeout, MMI read timeout, connection close timeout, ...). --- plc4go/internal/ads/Interactions.go | 2 +- plc4go/internal/bacnetip/Discoverer.go | 2 +- plc4go/internal/bacnetip/ReaderSegmentation.go | 2 +- plc4go/internal/bacnetip/SenderSegmentation.go | 3 +- plc4go/internal/cbus/Browser.go | 7 ++-- plc4go/internal/cbus/MessageCodec.go | 5 ++- plc4go/internal/eip/Connection.go | 6 +-- plc4go/internal/knxnetip/Connection.go | 5 ++- plc4go/internal/opcua/Connection.go | 7 ++-- plc4go/internal/simulated/Connection.go | 2 +- plc4go/pkg/api/cache/PlcConnectionCache.go | 5 ++- plc4go/pkg/api/cache/plcConnectionLease.go | 3 +- plc4go/spi/transactions/RequestTransaction.go | 3 +- plc4go/spi/utils/NamedTimeout.go | 42 +++++++++++++++++++++ plc4go/spi/utils/NamedTimeout_test.go | 52 ++++++++++++++++++++++++++ 15 files changed, 124 insertions(+), 22 deletions(-) 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 "<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)) +} 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") +}
