Skip to content

store: Don't hold write lock while receiving LTX #450

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

Merged
benbjohnson merged 3 commits into from
May 11, 2026
Merged

store: Don't hold write lock while receiving LTX #450

Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
62a68cc
store: Don't hold write lock while receiving LTX
saleemrashid Apr 21, 2026
81a86a5
store: Fix deadlock between AcquireRemoteHaltLock and processLTXStrea…
saleemrashid Apr 22, 2026
68dd622
mount: Fix race condition in WAL tests
saleemrashid Apr 22, 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
55 changes: 31 additions & 24 deletions store.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1522,14 +1522,6 @@ func (s *Store) processLTXStreamFrame(ctx context.Context, frame *LTXStreamFrame
TraceLog.Printf("[ProcessLTXStreamFrame.End(%s)]: %s", db.name, errorKeyValue(err))
}()

// Acquire lock unless we are waiting for a database position, in which case,
// we already have the lock.
guardSet, err := db.AcquireWriteLock(ctx, nil)
if err != nil {
return err
}
defer guardSet.Unlock()

// Skip frame if it already occurred on this node. This can happen if the
// replica node created the transaction and forwarded it to the primary.
if hdr.NodeID == s.ID() {
Expand All @@ -1543,24 +1535,12 @@ func (s *Store) processLTXStreamFrame(ctx context.Context, frame *LTXStreamFrame
return nil
}

// If we receive an LTX file while holding the remote HALT lock then the
// remote lock must have expired or been released so we can clear it locally.
//
// We also hold the local WRITE lock so a local write cannot be in-progress.
if haltLock := db.RemoteHaltLock(); haltLock != nil {
TraceLog.Printf("[ProcessLTXStreamFrame.Unhalt(%s)]: replica holds HALT lock but received LTX file, unsetting HALT lock", db.Name())
if err := db.UnsetRemoteHaltLock(ctx, haltLock.ID); err != nil {
return fmt.Errorf("release remote halt lock: %w", err)
}
}

// Verify LTX file pre-apply checksum matches the current database position
// unless this is a snapshot, which will overwrite all data.
// unless this is a snapshot, which will overwrite all data. This is a preflight
// check so we can skip streaming an LTX we can't apply, but we'll need to verify
// again after we acquire the write lock.
Copy link
Copy Markdown
Contributor Author

@saleemrashid saleemrashid Apr 21, 2026

Choose a reason for hiding this comment

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

There's a behavior change here now we're not acquiring the write lock: if a transaction is being applied concurrently, we might now observe the previous db.Pos() and return a position mismatch error here. In contrast, the existing code is serialized by the write lock, and would observe the post-apply db.Pos() of any transaction being applied concurrently.

I don't think this is an issue, but it is a behavior change, so I wanted to call it out.

if !hdr.IsSnapshot() {
expectedPos := ltx.Pos{
TXID: hdr.MinTXID - 1,
PostApplyChecksum: hdr.PreApplyChecksum,
}
expectedPos := hdr.PreApplyPos()
if pos := db.Pos(); pos != expectedPos {
return fmt.Errorf("position mismatch on db %q: %s <> %s", db.Name(), pos, expectedPos)
}
Expand Down Expand Up @@ -1595,6 +1575,33 @@ func (s *Store) processLTXStreamFrame(ctx context.Context, frame *LTXStreamFrame
dbLTXCountMetricVec.WithLabelValues(db.Name()).Inc()
dbLTXBytesMetricVec.WithLabelValues(db.Name()).Set(float64(n))

// Acquire lock unless we are waiting for a database position, in which case,
Comment thread
benbjohnson marked this conversation as resolved.
// we already have the lock.
guardSet, err := db.AcquireWriteLock(ctx, nil)
if err != nil {
return err
}
defer guardSet.Unlock()

// If we receive an LTX file while holding the remote HALT lock then the
// remote lock must have expired or been released so we can clear it locally.
//
// We also hold the local WRITE lock so a local write cannot be in-progress.
if haltLock := db.RemoteHaltLock(); haltLock != nil {
TraceLog.Printf("[ProcessLTXStreamFrame.Unhalt(%s)]: replica holds HALT lock but received LTX file, unsetting HALT lock", db.Name())
if err := db.UnsetRemoteHaltLock(ctx, haltLock.ID); err != nil {
return fmt.Errorf("release remote halt lock: %w", err)
}
}

// Verify the database position again, now we're holding the write lock.
if !hdr.IsSnapshot() {
expectedPos := hdr.PreApplyPos()
if pos := db.Pos(); pos != expectedPos {
return fmt.Errorf("position mismatch on db %q: %s <> %s", db.Name(), pos, expectedPos)
}
}

// Remove other LTX files after a snapshot.
if hdr.IsSnapshot() {
dir, file := filepath.Split(path)
Expand Down
Loading