From 5b8f4bb2dbf5d079ff91c95da392a802feaace63 Mon Sep 17 00:00:00 2001 From: Rahul Krishna Date: Sat, 5 Sep 2026 21:40:18 -0400 Subject: [PATCH] fix(reconcile): retry retryable Neo4j errors when creating constraints --- internal/reconcile/bolt.go | 30 ++-- internal/reconcile/bolt_constraint_test.go | 153 +++++++++++++++++++++ 2 files changed, 171 insertions(+), 12 deletions(-) create mode 100644 internal/reconcile/bolt_constraint_test.go diff --git a/internal/reconcile/bolt.go b/internal/reconcile/bolt.go index 65d7b28..0033f69 100644 --- a/internal/reconcile/bolt.go +++ b/internal/reconcile/bolt.go @@ -200,13 +200,17 @@ func (s *BoltStore) WriteGeneration(ctx context.Context, plan Plan) error { // createConstraints applies the catalog's schema statements in their own // session, because Neo4j refuses to mix schema and data in one transaction. +// Each statement gets its own managed transaction, so the driver retries the +// retryable errors — a deadlock between two writers creating the same +// `IF NOT EXISTS` constraint at once — with backoff instead of ending the +// generation. The statements are idempotent, so a retried attempt is safe. // -// A constraint failure is fatal to the generation rather than a warning: the -// uniqueness constraints are what make MERGE-by-id idempotent, and a shared -// graph really can refuse one (duplicate ids on a constrained label). Both the -// acknowledgement of RUN and the completion of the statement are checked, and -// the session's own close error is reported rather than dropped, so no failure -// can be lost between the two. +// A constraint failure that is not retryable stays fatal to the generation +// rather than a warning: the uniqueness constraints are what make MERGE-by-id +// idempotent, and a shared graph really can refuse one (duplicate ids on a +// constrained label). Both the acknowledgement of RUN and the completion of the +// statement are checked, and the session's own close error is reported rather +// than dropped, so no failure can be lost between the two. func (s *BoltStore) createConstraints(ctx context.Context, constraints []string) (err error) { session := s.session(ctx, neo4j.AccessModeWrite) defer func() { @@ -215,12 +219,14 @@ func (s *BoltStore) createConstraints(ctx context.Context, constraints []string) } }() for _, constraint := range constraints { - result, runErr := session.Run(ctx, constraint, nil) - if runErr != nil { - return fmt.Errorf("create constraint: %w", runErr) - } - if _, consumeErr := result.Consume(ctx); consumeErr != nil { - return fmt.Errorf("create constraint: %w", consumeErr) + if _, writeErr := session.ExecuteWrite(ctx, func(tx neo4j.ManagedTransaction) (any, error) { + result, runErr := tx.Run(ctx, constraint, nil) + if runErr != nil { + return nil, runErr + } + return result.Consume(ctx) + }); writeErr != nil { + return fmt.Errorf("create constraint: %w", writeErr) } } return nil diff --git a/internal/reconcile/bolt_constraint_test.go b/internal/reconcile/bolt_constraint_test.go new file mode 100644 index 0000000..c6139bc --- /dev/null +++ b/internal/reconcile/bolt_constraint_test.go @@ -0,0 +1,153 @@ +package reconcile + +import ( + "context" + "errors" + "testing" + + "github.com/neo4j/neo4j-go-driver/v5/neo4j" +) + +// transientDeadlock is the error Neo4j raises when two writers create the same +// constraint at once. Neo4j documents it as retryable, so the analyzer must not +// let it end a generation. +func transientDeadlock() error { + return &neo4j.Neo4jError{Code: "Neo.TransientError.Transaction.DeadlockDetected", Msg: "ForsetiClient[transactionId=51]"} +} + +// TestCreateConstraintsRetriesTransientError is the regression: one deadlock on +// the first attempt must be retried, not returned, and the constraint must +// exist once the retry lands. +func TestCreateConstraintsRetriesTransientError(t *testing.T) { + session := &fakeSession{errs: []error{transientDeadlock()}} + store := &BoltStore{driver: fakeDriver{session: session}} + + if err := store.createConstraints(context.Background(), []string{"CREATE CONSTRAINT c IF NOT EXISTS FOR (n:Artifact) REQUIRE n.id IS UNIQUE"}); err != nil { + t.Fatalf("createConstraints after one transient deadlock: %v", err) + } + if got, want := len(session.created), 1; got != want { + t.Fatalf("constraints created = %d, want %d", got, want) + } + if session.attempts != 2 { + t.Fatalf("attempts = %d, want 2 (one failure, one retry)", session.attempts) + } + if session.autocommit != 0 { + t.Fatalf("autocommit runs = %d, want 0: schema statements must go through a retrying managed transaction", session.autocommit) + } +} + +// TestCreateConstraintsFailsClosedOnNonRetryableError keeps the fail-closed +// guarantee: a real schema conflict still aborts the generation before any data +// is written, and is not retried. +func TestCreateConstraintsFailsClosedOnNonRetryableError(t *testing.T) { + refusal := &neo4j.Neo4jError{Code: "Neo.ClientError.Schema.ConstraintCreationFailed", Msg: "duplicate ids"} + session := &fakeSession{errs: []error{refusal}} + store := &BoltStore{driver: fakeDriver{session: session}} + + err := store.createConstraints(context.Background(), []string{"CREATE CONSTRAINT c IF NOT EXISTS FOR (n:Artifact) REQUIRE n.id IS UNIQUE"}) + var reported *neo4j.Neo4jError + if !errors.As(err, &reported) || reported.Code != refusal.Code { + t.Fatalf("createConstraints error = %v, want the driver's %s unchanged", err, refusal.Code) + } + if session.attempts != 1 { + t.Fatalf("attempts = %d, want 1: a non-retryable refusal must not be retried", session.attempts) + } + if len(session.created) != 0 { + t.Fatalf("constraints created = %d, want 0", len(session.created)) + } +} + +// TestCreateConstraintsStopsRetryingOnCancelledContext keeps a cancelled +// analysis from retrying a doomed statement forever. +func TestCreateConstraintsStopsRetryingOnCancelledContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + session := &fakeSession{errs: []error{transientDeadlock(), transientDeadlock(), transientDeadlock()}, cancel: cancel} + store := &BoltStore{driver: fakeDriver{session: session}} + + err := store.createConstraints(ctx, []string{"CREATE CONSTRAINT c IF NOT EXISTS FOR (n:Artifact) REQUIRE n.id IS UNIQUE"}) + if !errors.Is(err, context.Canceled) { + t.Fatalf("createConstraints error = %v, want context.Canceled", err) + } + if session.attempts != 1 { + t.Fatalf("attempts = %d, want 1: cancellation must stop the retries", session.attempts) + } +} + +// fakeDriver hands createConstraints one scripted session instead of a live +// connection, so the retry behaviour can be exercised without a server. +type fakeDriver struct { + neo4j.DriverWithContext + session neo4j.SessionWithContext +} + +func (d fakeDriver) NewSession(context.Context, neo4j.SessionConfig) neo4j.SessionWithContext { + return d.session +} + +// fakeSession scripts one error per attempt. ExecuteWrite stands in for the +// driver's managed transaction: it re-runs the unit of work while the error is +// retryable and the context is live, which is the behaviour the fix relies on. +// Run counts autocommit use, which is how the deadlock escaped in the first +// place. +type fakeSession struct { + neo4j.SessionWithContext + errs []error + attempts int + autocommit int + created []string + cancel context.CancelFunc +} + +func (s *fakeSession) attempt(cypher string) error { + var err error + if s.attempts < len(s.errs) { + err = s.errs[s.attempts] + } + s.attempts++ + if err == nil { + s.created = append(s.created, cypher) + } + if s.cancel != nil { + s.cancel() + } + return err +} + +func (s *fakeSession) Run(_ context.Context, cypher string, _ map[string]any, _ ...func(*neo4j.TransactionConfig)) (neo4j.ResultWithContext, error) { + s.autocommit++ + if err := s.attempt(cypher); err != nil { + return nil, err + } + return fakeResult{}, nil +} + +func (s *fakeSession) ExecuteWrite(ctx context.Context, work neo4j.ManagedTransactionWork, _ ...func(*neo4j.TransactionConfig)) (any, error) { + for { + if err := ctx.Err(); err != nil { + return nil, err + } + value, err := work(fakeTransaction{session: s}) + if err == nil || !neo4j.IsRetryable(err) { + return value, err + } + } +} + +func (s *fakeSession) Close(context.Context) error { return nil } + +type fakeTransaction struct { + neo4j.ManagedTransaction + session *fakeSession +} + +func (t fakeTransaction) Run(_ context.Context, cypher string, _ map[string]any) (neo4j.ResultWithContext, error) { + if err := t.session.attempt(cypher); err != nil { + return nil, err + } + return fakeResult{}, nil +} + +type fakeResult struct{ neo4j.ResultWithContext } + +func (fakeResult) Consume(context.Context) (neo4j.ResultSummary, error) { return nil, nil }