🗂️ SinkConsumerFlushedWalCursor #935
Open
+866
−172
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
RTLS
force-pushed
the
01-31-_sinkconsumerflushedwalcursor
branch
2 times, most recently
from
b97df28 to
1857743
Compare
RTLS
force-pushed
the
01-31-_sinkconsumerflushedwalcursor
branch
from
1857743 to
c6cb18d
Compare
RTLS
force-pushed
the
01-31-_sinkconsumerflushedwalcursor
branch
2 times, most recently
from
9fa7fb8 to
8fd0786
Compare
RTLS
force-pushed
the
01-31-_sinkconsumerflushedwalcursor
branch
from
8fd0786 to
836565c
Compare
acco
reviewed
Feb 12, 2025
| @@ -0,0 +1,49 @@ | |||
| defmodule Sequin.Consumers.SinkConsumerHighWatermarkWalCursor do | |||
| @@ -66,13 +67,14 @@ defmodule Sequin.ConsumersRuntime.ConsumerProducer do | |||
| @impl GenStage | |||
| def handle_info(:init, state) do | |||
| consumer = Repo.lazy_preload(state.consumer, postgres_database: [:replication_slot]) | |||
| {:ok, low_watermark} = Replication.find_watermark(consumer.replication_slot_id, :low) | |||
There was a problem hiding this comment.
what's my durable wal cursor?
Comment on lines
+199
to
+201
| less_than_low_watermark = | ||
| message.commit_lsn < state.low_watermark.commit_lsn or | ||
| (message.commit_lsn == state.low_watermark.commit_lsn and message.commit_idx < state.low_watermark.commit_idx) |
There was a problem hiding this comment.
incorrect, because failed messages can have <
| :ok | ||
|
|
||
| high_watermark_wal_cursor -> | ||
| {:ok, _} = |
Comment on lines
+124
to
+125
| @spec put_high_watermark_wal_cursor(State.t(), Replication.wal_cursor()) :: {:ok, State.t()} | ||
| def put_high_watermark_wal_cursor(%State{} = state, wal_cursor) do |
There was a problem hiding this comment.
slot_high_watermark
slot_non_durable_cursor
Comment on lines
+361
to
+362
| # Force update from the database so we advance the low watermark for acknowledgement | ||
| replication_slot = Repo.preload(state.replication_slot, [:low_watermark_wal_cursor], force: true) |
| messages = | ||
| if reply == 1 and not is_nil(state.last_commit_lsn) do | ||
| if reply == 1 and not is_nil(replication_slot.low_watermark_wal_cursor) do |
There was a problem hiding this comment.
Suggested change
| if reply == 1 and not is_nil(replication_slot.low_watermark_wal_cursor) do | |
| if reply == 1 do |
| @@ -730,7 +752,13 @@ defmodule Sequin.DatabasesRuntime.SlotProcessor do | |||
| send(state.test_pid, {__MODULE__, :heartbeat_received}) | |||
| end | |||
|
|
|||
| state | |||
| if accumulated_messages?(state) do | |||
RTLS
force-pushed
the
main
branch
3 times, most recently
from
6657098 to
459bc87
Compare
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
No description provided.