From e9379e66c0386d6ab94f21fc680027538ed3f4f2 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Thu, 27 Aug 2026 15:42:04 -0400 Subject: [PATCH 1/2] oracledb_cdc: make the mining backoff sleeps cancellable Review follow-on: ReadChanges backed off with plain time.Sleep, so a soft stop - Close's graceful path or the poisoned-publisher teardown in ReadBatch - had to wait out whatever mining or backoff interval was in flight before the session could report stopped. Both intervals are user-configurable and can be long, which undercut the teardown's bounded-wait contract. The sleeps now select on the session context and return its error, so cancellation cuts them short immediately. --- internal/impl/oracledb/logminer/logminer.go | 22 ++++++++++-- internal/impl/oracledb/logminer/sleep_test.go | 35 +++++++++++++++++++ 2 files changed, 55 insertions(+), 2 deletions(-) create mode 100644 internal/impl/oracledb/logminer/sleep_test.go diff --git a/internal/impl/oracledb/logminer/logminer.go b/internal/impl/oracledb/logminer/logminer.go index 5dd56b77aa..4771b72f8d 100644 --- a/internal/impl/oracledb/logminer/logminer.go +++ b/internal/impl/oracledb/logminer/logminer.go @@ -176,15 +176,33 @@ func (lm *LogMiner) ReadChanges(ctx context.Context, startPos replication.SCN) ( lm.log.Debugf("Caught up with redo logs, backing off for %s...", lm.cfg.MiningBackoffInterval) lm.caughtUpLogged = true } - time.Sleep(lm.cfg.MiningBackoffInterval) + if err := sleepInterruptible(ctx, lm.cfg.MiningBackoffInterval); err != nil { + return err + } } else { lm.caughtUpLogged = false - time.Sleep(lm.cfg.MiningInterval) + if err := sleepInterruptible(ctx, lm.cfg.MiningInterval); err != nil { + return err + } } } } } +// sleepInterruptible pauses for d unless ctx is cancelled first, so a +// shutdown or a poisoned-publisher teardown never has to wait out a +// mining backoff. +func sleepInterruptible(ctx context.Context, d time.Duration) error { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-timer.C: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + // FindStartPos returns the database's current SCN so that streaming begins from // the present moment rather than replaying historical redo logs. func (lm *LogMiner) FindStartPos(ctx context.Context) (replication.SCN, error) { diff --git a/internal/impl/oracledb/logminer/sleep_test.go b/internal/impl/oracledb/logminer/sleep_test.go new file mode 100644 index 0000000000..6cc103c6af --- /dev/null +++ b/internal/impl/oracledb/logminer/sleep_test.go @@ -0,0 +1,35 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md + +package logminer + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestSleepInterruptible(t *testing.T) { + t.Run("cancellation cuts the sleep short", func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + go func() { + time.Sleep(10 * time.Millisecond) + cancel() + }() + start := time.Now() + err := sleepInterruptible(ctx, time.Hour) + require.ErrorIs(t, err, context.Canceled) + require.Less(t, time.Since(start), 10*time.Second) + }) + + t.Run("uncancelled sleep elapses and returns nil", func(t *testing.T) { + require.NoError(t, sleepInterruptible(t.Context(), time.Millisecond)) + }) +} From d43d55be09869f96e6e1361b73421ed8749e5a19 Mon Sep 17 00:00:00 2001 From: Jonathan Chaput Date: Thu, 27 Aug 2026 15:42:04 -0400 Subject: [PATCH 2/2] mssqlserver_cdc: make the change-table backoff sleep cancellable Same exposure as the oracledb mining backoff: ReadChangeTables backed off with plain time.Sleep, so a soft stop - Close's graceful path or the poisoned-publisher teardown in ReadBatch - had to wait out the user-configurable backoff interval before the session could report stopped. The sleep now selects on the stream context and returns its error, so cancellation cuts it short immediately. --- .../mssqlserver/replication/sleep_test.go | 35 +++++++++++++++++++ .../impl/mssqlserver/replication/stream.go | 18 +++++++++- 2 files changed, 52 insertions(+), 1 deletion(-) create mode 100644 internal/impl/mssqlserver/replication/sleep_test.go diff --git a/internal/impl/mssqlserver/replication/sleep_test.go b/internal/impl/mssqlserver/replication/sleep_test.go new file mode 100644 index 0000000000..d4b97f9b7e --- /dev/null +++ b/internal/impl/mssqlserver/replication/sleep_test.go @@ -0,0 +1,35 @@ +// Copyright 2026 Redpanda Data, Inc. +// +// Licensed as a Redpanda Enterprise file under the Redpanda Community +// License (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// https://github.com/redpanda-data/connect/blob/main/licenses/rcl.md + +package replication + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestSleepInterruptible(t *testing.T) { + t.Run("cancellation cuts the sleep short", func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + go func() { + time.Sleep(10 * time.Millisecond) + cancel() + }() + start := time.Now() + err := sleepInterruptible(ctx, time.Hour) + require.ErrorIs(t, err, context.Canceled) + require.Less(t, time.Since(start), 10*time.Second) + }) + + t.Run("uncancelled sleep elapses and returns nil", func(t *testing.T) { + require.NoError(t, sleepInterruptible(t.Context(), time.Millisecond)) + }) +} diff --git a/internal/impl/mssqlserver/replication/stream.go b/internal/impl/mssqlserver/replication/stream.go index e53fa734a6..5f1772119f 100644 --- a/internal/impl/mssqlserver/replication/stream.go +++ b/internal/impl/mssqlserver/replication/stream.go @@ -496,12 +496,28 @@ func (r *ChangeTableStream) ReadChangeTables(ctx context.Context, db *sql.DB, st startLSN = lastLSN } else { r.log.Debug("No more changes across all change tables, backing off...") - time.Sleep(r.backoffInterval) + if err := sleepInterruptible(ctx, r.backoffInterval); err != nil { + return err + } } } } } +// sleepInterruptible pauses for d unless ctx is cancelled first, so a +// shutdown or a poisoned-publisher teardown never has to wait out the +// change-table backoff. +func sleepInterruptible(ctx context.Context, d time.Duration) error { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-timer.C: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + // UserDefinedTable represents a found user's SQL Server table (called a user-defined table) in SQL. type UserDefinedTable struct { Schema string