feat: aborted-transaction retry policy and client integration

client.go

@@ -87,6 +87,9 @@ type Client interface {
 	// DgraphClient returns a gRPC Dgraph client from the connection pool and a cleanup function.
 	// The cleanup function must be called when finished with the client to return it to the pool.
 	DgraphClient() (*dgo.Dgraph, func(), error)
+
+	// WithRetry executes fn, retrying on aborted transactions per policy.
+	WithRetry(ctx context.Context, policy RetryPolicy, fn func() error) error
 }
 
 const (

retry.go

@@ -0,0 +1,115 @@
+/*
+ * SPDX-FileCopyrightText: © 2017-2026 Istari Digital, Inc.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package modusgraph
+
+import (
+	"context"
+	"errors"
+	"math/rand/v2"
+	"time"
+
+	"github.com/dgraph-io/dgo/v250"
+)
+
+// RetryPolicy controls how WithRetry handles aborted transactions.
+// Modeled after dgraph4j's RetryPolicy: exponential backoff with jitter.
+type RetryPolicy struct {
+	// MaxRetries is the maximum number of retry attempts after the initial try.
+	MaxRetries int
+
+	// BaseDelay is the initial delay before the first retry.
+	// Subsequent delays grow exponentially: BaseDelay * 2^attempt.
+	BaseDelay time.Duration
+
+	// MaxDelay caps the backoff duration. No single delay exceeds this.
+	MaxDelay time.Duration
+
+	// Jitter adds randomness to each delay to prevent thundering herd.
+	// Expressed as a fraction of the computed delay (e.g. 0.1 = 10%).
+	Jitter float64
+}
+
+// DefaultRetryPolicy mirrors dgraph4j's defaults:
+// 5 retries, 100ms base delay, 5s max delay, 10% jitter.
+var DefaultRetryPolicy = RetryPolicy{
+	MaxRetries: 10,
+	BaseDelay:  100 * time.Millisecond,
+	MaxDelay:   5 * time.Second,
+	Jitter:     0.1,
+}
+
+// delay computes the backoff duration for a given attempt (0-indexed):
+// the exponential BaseDelay*2^attempt, plus up to Jitter of itself, clamped
+// so the result never exceeds MaxDelay. Clamping last keeps the documented
+// invariant that no single delay exceeds MaxDelay — adding jitter after the
+// cap would let the delay overshoot it.
+func (p RetryPolicy) delay(attempt int) time.Duration {
+	// Cap the exponential before jitter so a large attempt cannot overflow the
+	// shift. attempt comes from a range loop (>= 0) and is bounded below 63;
+	// exp <= 0 means the shift overflowed anyway, which we treat as the cap.
+	d := p.MaxDelay
+	if attempt >= 0 && attempt < 63 {
+		if exp := p.BaseDelay << attempt; exp > 0 && exp < p.MaxDelay {
+			d = exp
+		}
+	}
+	if p.Jitter > 0 {
+		// Backoff jitter spreads retriers apart; it does not need a
+		// cryptographic RNG, so math/rand/v2 is appropriate here.
+		d += time.Duration(float64(d) * p.Jitter * rand.Float64()) //nolint:gosec // G404: jitter is not security-sensitive
+		if d > p.MaxDelay {
+			d = p.MaxDelay
+		}
+	}
+	return d
+}
+
+// WithRetry executes fn, retrying on aborted transactions according to policy.
+//
+// This is an opt-in mechanism modeled after dgraph4j's client.withRetry().
+// The caller wraps their mutation logic in fn; WithRetry handles creating
+// fresh attempts with exponential backoff when Dgraph returns a transaction
+// abort due to concurrent conflicts.
+//
+// fn is called at least once. On each aborted-transaction error, WithRetry
+// waits according to the policy's backoff schedule and calls fn again, up to
+// policy.MaxRetries additional times. Non-abort errors are returned immediately.
+//
+// The context is checked between retries; if cancelled during a backoff sleep,
+// the context error is returned.
+//
+// Usage:
+//
+//	err := client.WithRetry(ctx, modusgraph.DefaultRetryPolicy, func() error {
+//	    return client.Insert(ctx, &entity)
+//	})
+func (c client) WithRetry(ctx context.Context, policy RetryPolicy, fn func() error) error {
+	// A negative MaxRetries would make the loop run zero times and never call
+	// fn; clamp to zero so fn always runs at least once, as documented.
+	maxRetries := policy.MaxRetries
+	if maxRetries < 0 {
+		maxRetries = 0
+	}
+	for attempt := range maxRetries + 1 {
+		err := fn()
+		if err == nil {
+			return nil
+		}
+		if !errors.Is(err, dgo.ErrAborted) || attempt >= maxRetries {
+			return err
+		}
+		d := policy.delay(attempt)
+		c.logger.V(1).Info("Transaction aborted, retrying",
+			"attempt", attempt+1, "maxRetries", maxRetries, "delay", d)
+		select {
+		case <-time.After(d):
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+	// Unreachable: the loop runs at least once and returns on every path.
+	panic("unreachable")
+}

retry_example_test.go

@@ -0,0 +1,53 @@
+/*
+ * SPDX-FileCopyrightText: © 2017-2026 Istari Digital, Inc.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package modusgraph_test
+
+import (
+	"context"
+	"time"
+
+	"github.com/matthewmcneely/modusgraph"
+)
+
+// ExampleClient_WithRetry wraps a mutation so an aborted transaction — the
+// error Dgraph returns when concurrent writers conflict on an indexed
+// predicate — is retried with exponential backoff instead of surfacing to the
+// caller. Non-abort errors return immediately; the context bounds the total
+// wait.
+func ExampleClient_withRetry() {
+	client, _ := modusgraph.NewClient("dgraph://localhost:9080")
+	defer client.Close()
+
+	ctx := context.Background()
+	entity := &RetryEntity{Name: "alice", Value: 1}
+
+	err := client.WithRetry(ctx, modusgraph.DefaultRetryPolicy, func() error {
+		return client.Insert(ctx, entity)
+	})
+	if err != nil {
+		panic(err)
+	}
+}
+
+// ExampleRetryPolicy shows a custom backoff schedule: three retries, 50ms base
+// delay doubling each attempt, capped at 2s, with 20% jitter to spread
+// concurrent retriers apart.
+func ExampleRetryPolicy() {
+	client, _ := modusgraph.NewClient("dgraph://localhost:9080")
+	defer client.Close()
+
+	policy := modusgraph.RetryPolicy{
+		MaxRetries: 3,
+		BaseDelay:  50 * time.Millisecond,
+		MaxDelay:   2 * time.Second,
+		Jitter:     0.2,
+	}
+
+	ctx := context.Background()
+	_ = client.WithRetry(ctx, policy, func() error {
+		return client.Insert(ctx, &RetryEntity{Name: "bob", Value: 2})
+	})
+}

retry_internal_test.go

@@ -0,0 +1,88 @@
+/*
+ * SPDX-FileCopyrightText: © 2017-2026 Istari Digital, Inc.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package modusgraph
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestRetryPolicyDelayExponentialGrowth(t *testing.T) {
+	p := RetryPolicy{
+		BaseDelay: 100 * time.Millisecond,
+		MaxDelay:  10 * time.Second,
+		Jitter:    0,
+	}
+
+	assert.Equal(t, 100*time.Millisecond, p.delay(0))
+	assert.Equal(t, 200*time.Millisecond, p.delay(1))
+	assert.Equal(t, 400*time.Millisecond, p.delay(2))
+	assert.Equal(t, 800*time.Millisecond, p.delay(3))
+	assert.Equal(t, 1600*time.Millisecond, p.delay(4))
+}
+
+func TestRetryPolicyDelayMaxCap(t *testing.T) {
+	p := RetryPolicy{
+		BaseDelay: 1 * time.Second,
+		MaxDelay:  3 * time.Second,
+		Jitter:    0,
+	}
+
+	assert.Equal(t, 1*time.Second, p.delay(0))
+	assert.Equal(t, 2*time.Second, p.delay(1))
+	assert.Equal(t, 3*time.Second, p.delay(2))
+	assert.Equal(t, 3*time.Second, p.delay(3))
+	assert.Equal(t, 3*time.Second, p.delay(10))
+}
+
+func TestRetryPolicyDelayWithJitter(t *testing.T) {
+	p := RetryPolicy{
+		BaseDelay: 100 * time.Millisecond,
+		MaxDelay:  10 * time.Second,
+		Jitter:    0.5,
+	}
+
+	for range 100 {
+		d := p.delay(0)
+		assert.GreaterOrEqual(t, d, 100*time.Millisecond, "delay should be at least base")
+		assert.LessOrEqual(t, d, 150*time.Millisecond, "delay should not exceed base + 50% jitter")
+	}
+}
+
+// TestRetryPolicyDelayJitterNeverExceedsMaxCap pins the documented invariant:
+// even when the exponential delay sits at or above MaxDelay and jitter is
+// large, no single delay exceeds MaxDelay. Jitter is added before the final
+// clamp, so the result stays within the cap.
+func TestRetryPolicyDelayJitterNeverExceedsMaxCap(t *testing.T) {
+	p := RetryPolicy{
+		BaseDelay: 8 * time.Second,
+		MaxDelay:  10 * time.Second,
+		Jitter:    0.5, // up to +50% would overshoot MaxDelay without the clamp
+	}
+	for attempt := range 6 {
+		for range 100 {
+			d := p.delay(attempt)
+			assert.LessOrEqual(t, d, p.MaxDelay,
+				"attempt %d: delay %v exceeded MaxDelay %v", attempt, d, p.MaxDelay)
+			assert.Positive(t, d, "attempt %d: delay should be positive", attempt)
+		}
+	}
+}
+
+func TestRetryPolicyDelayZeroJitter(t *testing.T) {
+	p := RetryPolicy{
+		BaseDelay: 100 * time.Millisecond,
+		MaxDelay:  10 * time.Second,
+		Jitter:    0,
+	}
+
+	for range 10 {
+		assert.Equal(t, 100*time.Millisecond, p.delay(0))
+		assert.Equal(t, 200*time.Millisecond, p.delay(1))
+	}
+}