Skip to content

Mutex in tx monitor #264

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
GabrielMartinezRodriguez merged 4 commits into from
Dec 27, 2024
Merged

Mutex in tx monitor #264

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
64ec928
feat(txm): mutex in tx monitor
GabrielMartinezRodriguez Nov 20, 2024
682b147
chore: pr comments
GabrielMartinezRodriguez Nov 21, 2024
4ae447b
chore(txm): refactor mutex
GabrielMartinezRodriguez Dec 20, 2024
89f8e7e
chore(txm): remove unused dependency
GabrielMartinezRodriguez Dec 23, 2024
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
26 changes: 26 additions & 0 deletions packages/transaction-manager/lib/TxMonitor.ts
View file
Open in desktop

@norswap norswap Dec 5, 2024 •
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.

Maybe I'm being too left curve, but couldn't we just have a field private locked = false

And then in onNewBlock:

if (this.locked) return
this.locked = true
// ....
this.locked = false // probably in a finally statement though

Seems a lot simpler, more performant, and one less dependency :)

@GabrielMartinezRodriguez GabrielMartinezRodriguez Dec 6, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Yes, we could. The main difference is that, in your case, the TXM would have to wait until a new block arrives. With the mutex that I developed, as soon as the monitoring finishes, the last pending block starts immediately.

For example, if the block monitor for block N takes 2.1 seconds, in my case, block N+1 will start being monitored immediately after the finalization of block N. In your case, you would lose block N+1 and would have to wait until the reception of block N+2

@norswap norswap Dec 20, 2024

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.

Okay that makes sense.

@norswap norswap Dec 20, 2024

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.

The solution is perfectly fine, but there is something with the mutex code that fucks with my brain (all the indirection most like).

Attempt 1 at a simpler solution:

    private locked = false
    private pendingBlock: LatestBlock | undefined = undefined

    private onNewBlock(block: LatestBlock) {
        if (this.locked) {
            this.pendingBlock = block
        } else {
            this.locked = true
            this.handleNewBlock(block)
                .catch((error) => {
                    console.error("Error in handleNewBlock: ", error)
                })
                .finally(() => {
                    this.locked = false
                    if (this.pendingBlock) {
                        const block = this.pendingBlock
                        this.pendingBlock = undefined
                        this.onNewBlock(block)
                    }
                })
        }
    }

I like this, but it has a flaw: if we're systematically lagging, we'll end up recursing and trigger a stack too deep.

Attempt 2 doesn't have this issue and should behave the same as your proposal. It's a tad longer than attempt 1 but imho it reads even cleaner:

    private locked = false
    private pendingBlockPromises: PromiseWithResolvers<void>[] = []

    private async onNewBlock(block: LatestBlock) {
        if (this.locked) {
            const pending = promiseWithResolvers<void>()
            this.pendingBlockPromises.push(pending)
            try {
                await pending.promise
            } catch {
                // A more recent block came while we were waiting, abort.
                return
            }
        }

        this.locked = true
        try {
            await this.handleNewBlock(block)
        } catch (error) {
            console.error("Error in handleNewBlock: ", error)
        }
        this.locked = false

        this.pendingBlockPromises.pop()?.resolve()
        this.pendingBlockPromises.forEach((p) => p.reject())
    }

Let me know what you think (DM me, since there's no easy way for me to discriminate between PR comment notificatons).

@GabrielMartinezRodriguez GabrielMartinezRodriguez Dec 20, 2024

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I like your solution

Original file line number Diff line number Diff line change
Expand Up @@ -10,13 +10,39 @@ type AttemptWithReceipt = { attempt: Attempt; receipt: TransactionReceipt }

export class TxMonitor {
private readonly transactionManager: TransactionManager
private locked = false
private pendingBlockPromises: PromiseWithResolvers<void>[] = []

constructor(transactionManager: TransactionManager) {
this.transactionManager = transactionManager
eventBus.on(Topics.NewBlock, this.onNewBlock.bind(this))
}

private async onNewBlock(block: LatestBlock) {
if (this.locked) {
const pending = promiseWithResolvers<void>()
this.pendingBlockPromises.push(pending)
try {
await pending.promise
} catch {
// A more recent block came while we were waiting, abort.
return
}
}

this.locked = true
try {
await this.handleNewBlock(block)
} catch (error) {
console.error("Error in handleNewBlock: ", error)
}
this.locked = false

this.pendingBlockPromises.pop()?.resolve()
this.pendingBlockPromises.forEach((p) => p.reject())
}

private async handleNewBlock(block: LatestBlock) {
const transactions = this.transactionManager.transactionRepository.getNotFinalizedTransactions()

const promises = transactions.map(async (transaction) => {
Expand Down