Skip to content

fix: move RetryCount default to config.Validate() across all drivers #1058

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cotishq wants to merge 14 commits into
base: staging
Choose a base branch
from
Open

fix: move RetryCount default to config.Validate() across all drivers #1058

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
861adcb
fix: move RetryCount default to config.Validate() across all drivers
cotishq Jul 31, 2026
8b34343
fix: return RetryCount + 1 from MaxRetries to align with retry count …
cotishq Aug 5, 2026
f3b2014
fix: centralize +1 retry count handling in RetryOnBackoff
cotishq Aug 19, 2026
595d408
Merge branch 'staging' into fix/retry-count-initialization
saksham-datazip Aug 21, 2026
48bb82b
Merge branch 'staging' into fix/retry-count-initialization
saksham-datazip Aug 21, 2026
38f3ef9
fix: move RetryCount default to config.Validate() across all drivers
cotishq Jul 31, 2026
4b6e20f
fix: return RetryCount + 1 from MaxRetries to align with retry count …
cotishq Aug 5, 2026
75336a9
fix: centralize +1 retry count handling in RetryOnBackoff
cotishq Aug 19, 2026
f0b4a29
fix: fail sync on negative RetryCount and centralize retry logic
cotishq Aug 21, 2026
5b0e0c6
Merge branch 'fix/retry-count-initialization' of github.com:cotishq/o…
cotishq Aug 21, 2026
4d1139d
style: clean up RetryOnBackoff doc comment
cotishq Aug 21, 2026
d366f74
Merge branch 'staging' into fix/retry-count-initialization
saksham-datazip Aug 26, 2026
3c2bae8
Merge branch 'staging' into fix/retry-count-initialization
cotishq Sep 1, 2026
9a754e2
fix: restore mysql database default and align MaxThreads validation
cotishq Sep 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion drivers/db2/internal/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,10 +79,22 @@ func (c *Config) Validate() error {
return fmt.Errorf("database name is required")
}

if c.MaxThreads <= 0 {
if c.MaxThreads < 0 {
return fmt.Errorf("max threads is required")
}

if c.MaxThreads == 0 {
c.MaxThreads = constants.DefaultThreadCount
}

if c.RetryCount < 0 {
return fmt.Errorf("retry count is required")
}

if c.RetryCount == 0 {
c.RetryCount = constants.DefaultRetryCount
Comment on lines +91 to +95

@saksham-datazip saksham-datazip Aug 26, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you please add a space between two if conditions and do it for all the drivers wherever needed?

}

if c.SSLConfiguration == nil {
c.SSLConfiguration = &utils.SSLConfig{
Mode: "disable",
Expand Down
26 changes: 26 additions & 0 deletions drivers/db2/internal/config_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,32 @@ func TestConfig_Validate(t *testing.T) {
},
expectErr: true,
},
{
// Rejects config with a negative max threads.
name: "invalid config - negative max threads",
config: &Config{
Host: "db2-host",
Port: 50000,
Database: "testdb",
Username: "db2inst1",
Password: "secret1234",
MaxThreads: -1,
},
expectErr: true,
},
{
// Rejects config with a negative retry count.
name: "invalid config - negative retry count",
config: &Config{
Host: "db2-host",
Port: 50000,
Database: "testdb",
Username: "db2inst1",
Password: "secret1234",
RetryCount: -1,
},
expectErr: true,
},
}

for _, tt := range tests {
Expand Down
2 changes: 0 additions & 2 deletions drivers/db2/internal/db2.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@ import (
"github.com/datazip-inc/olake/drivers/abstract"
"github.com/datazip-inc/olake/pkg/jdbc"
"github.com/datazip-inc/olake/types"
"github.com/datazip-inc/olake/utils"
"github.com/datazip-inc/olake/utils/logger"
"github.com/jmoiron/sqlx"
"golang.org/x/crypto/ssh"
Expand Down Expand Up @@ -79,7 +78,6 @@ func (d *DB2) Setup(ctx context.Context) error {
}

d.client = client
d.config.RetryCount = utils.Ternary(d.config.RetryCount <= 0, 1, d.config.RetryCount+1).(int)
return nil
Comment on lines -82 to 81

@saksham-datazip saksham-datazip Aug 5, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In olake , by backoff_retry_count we mean is number of retry count for example if user enters 3 backoff retry count than it means 1 main sync + 3 Retry count so can you please make changes according to that.

@cotishq cotishq Aug 5, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeahh that makes sense, i have updated all drivers to align with this semantic ( attempts = RetryCount + 1 )

please lmk if this looks good

}

Expand Down
12 changes: 10 additions & 2 deletions drivers/kafka/internal/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,11 +65,19 @@ func (c *Config) Validate() error {
}
}

if c.MaxThreads <= 0 {
if c.MaxThreads < 0 {
return fmt.Errorf("max threads is required")
}

if c.MaxThreads == 0 {
c.MaxThreads = constants.DefaultThreadCount
}

if c.RetryCount <= 0 {
if c.RetryCount < 0 {
return fmt.Errorf("retry count is required")
}

if c.RetryCount == 0 {
c.RetryCount = constants.DefaultRetryCount
}

Expand Down
4 changes: 0 additions & 4 deletions drivers/kafka/internal/kafka.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -130,10 +130,6 @@ func (k *Kafka) Setup(ctx context.Context) error {
logger.Infof("initialized schema registry client for endpoint: %s", k.config.SchemaRegistry.Endpoint)
}

// TODO: Avoid modifying the configured/default retry count during initialization across all drivers.
// check for default backoff count
k.config.RetryCount = utils.Ternary(k.config.RetryCount <= 0, 1, k.config.RetryCount+1).(int)

return nil
}

Expand Down
2 changes: 1 addition & 1 deletion drivers/mongodb/internal/backfill.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ func (m *Mongo) GetOrSplitChunks(ctx context.Context, pool *destination.WriterPo
// Generate and update chunks
var retryErr error
var chunksArray []types.Chunk
err = utils.RetryOnBackoff(ctx, m.config.RetryCount, constants.DefaultRetryTimeout, func(ctx context.Context) error {
err = utils.RetryOnBackoff(ctx, m.MaxRetries(), constants.DefaultRetryTimeout, func(ctx context.Context) error {
chunksArray, retryErr = m.splitChunks(ctx, collection, stream, storageSize)
return retryErr
})
Expand Down
12 changes: 10 additions & 2 deletions drivers/mongodb/internal/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,11 +112,19 @@ func (c *Config) Validate() error {
// MongoDB rejects at connect time if the mechanism actually needs a password.
}

if c.MaxThreads <= 0 {
if c.MaxThreads < 0 {
return fmt.Errorf("max threads is required")
}

if c.MaxThreads == 0 {
c.MaxThreads = constants.DefaultThreadCount
}

if c.RetryCount <= 0 {
if c.RetryCount < 0 {
return fmt.Errorf("retry count is required")
}

if c.RetryCount == 0 {
c.RetryCount = constants.DefaultRetryCount
}

Expand Down
24 changes: 24 additions & 0 deletions drivers/mongodb/internal/config_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -186,6 +186,30 @@ func TestConfig_Validate(t *testing.T) {
},
expectErr: true,
},
{
name: "negative max threads fails validation",
config: &Config{
Hosts: []string{"mongo.example.com:27017"},
Database: "testdb",
Username: "user",
Password: "pass",
AuthDB: "admin",
MaxThreads: -1,
},
expectErr: true,
},
{
name: "negative retry count fails validation",
config: &Config{
Hosts: []string{"mongo.example.com:27017"},
Database: "testdb",
Username: "user",
Password: "pass",
AuthDB: "admin",
RetryCount: -1,
},
expectErr: true,
},
{
name: "sets defaults",
config: &Config{
Expand Down
2 changes: 0 additions & 2 deletions drivers/mongodb/internal/mon.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,8 +171,6 @@ func (m *Mongo) Setup(ctx context.Context) error {
m.client = conn
// no need to check from discover if it have cdc support or not
m.CDCSupport = true
// check for default backoff count
m.config.RetryCount = utils.Ternary(m.config.RetryCount == 0, 1, m.config.RetryCount+1).(int)
pingCtx, cancel := context.WithTimeout(ctx, 1*time.Minute)
Comment on lines 173 to 174

@saksham-datazip saksham-datazip Aug 5, 2026 •
edited
Loading

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So what i was suggesting is instead of removing this part you can let it stay as it is and just remove that check from validate function so it would be single source of truth and just add a small comment over here specifying why we are doing +1 .
if you have some other reasoning for the changes you made than feel free to discuss.

@cotishq cotishq Aug 5, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok, i actually agree the +1 is intentional, my pr doesnt remove it, it relocates it, retry_count is the no. of retries in RetryOnBackoff expects total attempts, so Maxretries now returns retrycount + 1 ( same behaviour btw ) but the conversation happens at the interface boundary instead of mutating config,

and on single source of truth: the line was copy-pasted into 9 drivers setup() method. after my change it exists once i.e constants. DefaultRetryCount + Validate().

also, keeping it in Setup() cant fix the actual bug cleanly imo, like the unset retry_count became 1 attempt (zero retries). That fix requires an inline DefaultRetryCount+1 inside the Ternary, which is exactly what needs the comment you're asking for. In Validate() it reads naturally, and it follows the existing MaxThreads precedent in the same file

i think the middle ground would be : i'll add the comment you suggested at the conversion point in MaxRetries() , like

// RetryCount is the number of retries; +1 accounts for the initial attempt,
// since RetryOnBackoff expects total attempts.
return p.config.RetryCount + 1

lmk if that works

@saksham-datazip saksham-datazip Aug 18, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I understand what you mean. I think we can handle this in the Setup() function itself: if RetryCount < 0, we fail the sync, and if RetryCount == 0, we set it to the default retry count.

if c.RetryCount < 0 {
    return fmt.Errorf("retry count is required")
}

if c.RetryCount == 0 {
    c.RetryCount = constants.DefaultRetryCount
}

Apart from that, instead of modifying the value in MaxRetries, we can handle the +1 in a single place inside RetryOnBackoff() and add a comment explaining why it is needed:

func RetryOnBackoff(ctx context.Context, attempts int, sleep time.Duration, f func(ctx context.Context) error) (err error) {
    // Add 1 because attempts represents the number of retries,
    // while the function needs the total number of attempts.
    attempts = attempts + 1

This way, we are not manipulating the input provided by the user in multiple places, and the retry-count logic stays centralized in RetryOnBackoff().

And just FYI, in your approach, Mongo is directly calling RetryOnBackoff like this:

err = utils.RetryOnBackoff(ctx, m.config.RetryCount, constants.DefaultRetryTimeout, func(ctx context.Context) error {
    chunksArray, retryErr = m.splitChunks(ctx, collection, stream, storageSize)
    return retryErr
})

So it would bypass the changes you're making in MaxRetries and use default values.

@saksham-datazip saksham-datazip Aug 18, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also their is one more thing we have changed the scope of this pr that we dont want config file to be manipulated directly so can you once check in entire pr that what we should do for instance in mysql if db name is empty we are setting it to "mysql" but we should fail the sync itself can you once check and made the respective changes

defer cancel()

Expand Down
12 changes: 10 additions & 2 deletions drivers/mssql/internal/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,11 +51,19 @@ func (c *Config) Validate() error {
return fmt.Errorf("database is required")
}

if c.MaxThreads <= 0 {
if c.MaxThreads < 0 {
return fmt.Errorf("max threads is required")
}

if c.MaxThreads == 0 {
c.MaxThreads = constants.DefaultThreadCount
}

if c.RetryCount <= 0 {
if c.RetryCount < 0 {
return fmt.Errorf("retry count is required")
}

if c.RetryCount == 0 {
c.RetryCount = constants.DefaultRetryCount
}

Expand Down
26 changes: 26 additions & 0 deletions drivers/mssql/internal/config_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -136,6 +136,32 @@ func TestConfig_Validate(t *testing.T) {
},
expectErr: true,
},
{
// Rejects config with a negative max threads.
name: "invalid config - negative max threads",
config: &Config{
Host: "mssql-host",
Port: 1433,
Database: "testdb",
Username: "sa",
Password: "Password!123",
MaxThreads: -1,
},
expectErr: true,
},
{
// Rejects config with a negative retry count.
name: "invalid config - negative retry count",
config: &Config{
Host: "mssql-host",
Port: 1433,
Database: "testdb",
Username: "sa",
Password: "Password!123",
RetryCount: -1,
},
expectErr: true,
},
{
name: "valid config with primary_config",
config: &Config{
Expand Down
1 change: 0 additions & 1 deletion drivers/mssql/internal/mssql.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,6 @@ func (m *MSSQL) Setup(ctx context.Context) error {
return fmt.Errorf("failed to connect to MSSQL: %w", err)
}

m.config.RetryCount = utils.Ternary(m.config.RetryCount <= 0, 1, m.config.RetryCount+1).(int)
// Enable CDC support if database-level CDC is enabled
cdcSupported, err := m.isDatabaseCDCEnabled(ctx)
if err != nil {
Expand Down
12 changes: 10 additions & 2 deletions drivers/mysql/internal/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,12 +113,20 @@ func (c *Config) Validate() error {
}

// Set default number of threads if not provided
if c.MaxThreads <= 0 {
if c.MaxThreads < 0 {
return fmt.Errorf("max threads is required")
}

if c.MaxThreads == 0 {
c.MaxThreads = constants.DefaultThreadCount // Aligned with PostgreSQL default
}

// Set default retry count if not provided
if c.RetryCount <= 0 {
if c.RetryCount < 0 {
return fmt.Errorf("retry count is required")
}

if c.RetryCount == 0 {
c.RetryCount = constants.DefaultRetryCount // Reasonable default for retries
}

Expand Down
30 changes: 28 additions & 2 deletions drivers/mysql/internal/config_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -337,6 +337,32 @@ func TestConfig_Validate(t *testing.T) {
},
expectErr: false,
},
// Rejects negative max threads.
{
name: "invalid config - negative max threads",
config: &Config{
Host: "localhost",
Port: 3306,
Username: "testuser",
Password: "testpass",
Database: "testdb",
MaxThreads: -1,
},
expectErr: true,
},
// Rejects negative retry count.
{
name: "invalid config - negative retry count",
config: &Config{
Host: "localhost",
Port: 3306,
Username: "testuser",
Password: "testpass",
Database: "testdb",
RetryCount: -1,
},
expectErr: true,
},
}

for _, tt := range tests {
Expand All @@ -348,8 +374,8 @@ func TestConfig_Validate(t *testing.T) {
if !tt.expectErr && err != nil {
t.Errorf("Expected no error but got: %v", err)
}
if !tt.expectErr && err == nil && tt.config.Database == "" {
t.Errorf("Expected database to default to 'mysql'")
if !tt.expectErr && err == nil && tt.name == "valid config - defaults database" && tt.config.Database != "mysql" {
t.Errorf("Expected database to default to 'mysql', got %q", tt.config.Database)
}
})
}
Expand Down
1 change: 0 additions & 1 deletion drivers/mysql/internal/mysql.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,7 +154,6 @@ func (m *MySQL) Setup(ctx context.Context) error {
m.cdcConfig = *cdc
}
m.client = client
m.config.RetryCount = utils.Ternary(m.config.RetryCount <= 0, 1, m.config.RetryCount+1).(int)
// Enable CDC support if binlog is configured
cdcSupported, err := m.IsCDCSupported(ctx)
if err != nil {
Expand Down
14 changes: 13 additions & 1 deletion drivers/oracle/internal/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,10 +72,22 @@ func (c *Config) Validate() error {
}

// Set default number of threads if not provided
if c.MaxThreads <= 0 {
if c.MaxThreads < 0 {
return fmt.Errorf("max threads is required")
}

if c.MaxThreads == 0 {
c.MaxThreads = constants.DefaultThreadCount
}

if c.RetryCount < 0 {
return fmt.Errorf("retry count is required")
}

if c.RetryCount == 0 {
c.RetryCount = constants.DefaultRetryCount
}

if c.SSLConfiguration == nil {
c.SSLConfiguration = &utils.SSLConfig{
Mode: "disable",
Expand Down
1 change: 0 additions & 1 deletion drivers/oracle/internal/oracle.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,6 @@ func (o *Oracle) Setup(ctx context.Context) error {
}

o.client = client
o.config.RetryCount = utils.Ternary(o.config.RetryCount <= 0, 1, o.config.RetryCount+1).(int)
return nil
}

Expand Down
19 changes: 18 additions & 1 deletion drivers/postgres/internal/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,11 +47,28 @@ func (c *Config) Validate() error {
return fmt.Errorf("invalid port number: must be between 1 and 65535")
}

if c.Database == "" {
return fmt.Errorf("database name is required")
}

// default number of threads
if c.MaxThreads <= 0 {
if c.MaxThreads < 0 {
return fmt.Errorf("max threads is required")
}

if c.MaxThreads == 0 {
c.MaxThreads = constants.DefaultThreadCount
}

// default backoff retry count
if c.RetryCount < 0 {
return fmt.Errorf("retry count is required")
}

if c.RetryCount == 0 {
c.RetryCount = constants.DefaultRetryCount
}

// Add the connection parameters to the url
parsed := &url.URL{
Scheme: "postgres",
Expand Down
Loading