From 3972ae0b4e0ac9082060e8be45324672c652b440 Mon Sep 17 00:00:00 2001 From: "cursor[bot]" <206951365+cursor[bot]@users.noreply.github.com> Date: Mon, 22 Jun 2026 12:36:18 +0530 Subject: [PATCH] Read pending scheduled posts from master to avoid replica lag gaps (#37104) * Fix scheduled post job to fetch pending posts from master The scheduled post job fetches a page of pending posts, deletes the processed ones, then fetches the next page. Reading the page from a read replica can return stale data, causing already-processed posts to reappear on later pages. Read from master to avoid this gap. Co-authored-by: mattermost-code * Add regression test for scheduled post master read Verifies GetPendingScheduledPosts reads pending posts from the master database by pointing the replica at a separate empty database; a read from the replica would return no rows. Co-authored-by: mattermost-code --------- Co-authored-by: Cursor Agent Co-authored-by: mattermost-code --- .../store/sqlstore/scheduled_post_store.go | 5 +- .../sqlstore/scheduled_post_store_test.go | 71 +++++++++++++++++++ 2 files changed, 75 insertions(+), 1 deletion(-) diff --git a/server/channels/store/sqlstore/scheduled_post_store.go b/server/channels/store/sqlstore/scheduled_post_store.go index a5bc212fb19..3970e6e9fbe 100644 --- a/server/channels/store/sqlstore/scheduled_post_store.go +++ b/server/channels/store/sqlstore/scheduled_post_store.go @@ -179,8 +179,11 @@ func (s *SqlScheduledPostStore) GetPendingScheduledPosts(beforeTime, afterTime i }) } + // We read from the master here instead of a replica on purpose. The scheduled post job + // deletes processed posts and then fetches the next page of pending posts. Reading from a + // replica can return stale data, causing already-processed posts to reappear on later pages. var scheduledPosts []*model.ScheduledPost - if err := s.GetReplica().SelectBuilder(&scheduledPosts, query); err != nil { + if err := s.GetMaster().SelectBuilder(&scheduledPosts, query); err != nil { mlog.Error( "SqlScheduledPostStore.GetPendingScheduledPosts: failed to fetch pending scheduled posts for processing", mlog.Int("before_time", beforeTime), diff --git a/server/channels/store/sqlstore/scheduled_post_store_test.go b/server/channels/store/sqlstore/scheduled_post_store_test.go index 7365982d061..50ea5f29cfc 100644 --- a/server/channels/store/sqlstore/scheduled_post_store_test.go +++ b/server/channels/store/sqlstore/scheduled_post_store_test.go @@ -6,9 +6,80 @@ package sqlstore import ( "testing" + "github.com/stretchr/testify/require" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/shared/mlog" "github.com/mattermost/mattermost/server/v8/channels/store/storetest" ) func TestScheduledPostStore(t *testing.T) { StoreTestWithSqlStore(t, storetest.TestScheduledPostStore) } + +// TestGetPendingScheduledPostsReadsFromMaster verifies that the scheduled post job reads its +// pending posts from the master database rather than a read replica. The job deletes processed +// posts and then fetches the next page, so reading from a lagging replica could surface +// already-processed posts again. The test points the replica at a separate, empty database so +// that a read from the replica would return nothing, proving the read targets master. +func TestGetPendingScheduledPostsReadsFromMaster(t *testing.T) { + if testing.Short() { + t.Skip("skipping test in short mode") + } + + logger := mlog.CreateTestLogger(t) + + masterSettings, err := makeSqlSettings(model.DatabaseDriverPostgres) + if err != nil { + t.Skip(err) + } + defer storetest.CleanupSqlSettings(masterSettings) + + // The replica lives in its own database. We migrate it so the ScheduledPosts table exists + // but leave it empty: the row under test will only ever exist on master. + replicaSettings, err := makeSqlSettings(model.DatabaseDriverPostgres) + if err != nil { + t.Skip(err) + } + defer storetest.CleanupSqlSettings(replicaSettings) + + replicaStore, err := New(*replicaSettings, logger, nil) + require.NoError(t, err) + replicaStore.Close() + + masterSettings.DataSourceReplicas = []string{*replicaSettings.DataSource} + + store, err := New(*masterSettings, logger, nil) + require.NoError(t, err) + defer store.Close() + + // A license is required for replica reads to route to the replica. Without one, GetReplica() + // falls back to master and the test could pass even if the code read from the replica. + store.UpdateLicense(&model.License{}) + require.NotSame(t, store.GetMaster(), store.GetReplica(), "replica must be a distinct connection for this test to be meaningful") + + scheduledPost := &model.ScheduledPost{ + Draft: model.Draft{ + CreateAt: model.GetMillis(), + UserId: model.NewId(), + ChannelId: model.NewId(), + Message: "pending scheduled post", + }, + ScheduledAt: model.GetMillis(), + } + createdScheduledPost, err := store.ScheduledPost().CreateScheduledPost(scheduledPost) + require.NoError(t, err) + require.NotEmpty(t, createdScheduledPost.Id) + + // Sanity check: the replica database does not contain the scheduled post. + var replicaCount int + require.NoError(t, store.GetReplica().Get(&replicaCount, "SELECT COUNT(*) FROM ScheduledPosts")) + require.Zero(t, replicaCount, "replica should be empty so the test can distinguish master from replica reads") + + beforeTime := createdScheduledPost.ScheduledAt + 1000 + afterTime := createdScheduledPost.ScheduledAt - (24 * 60 * 60 * 1000) + pending, err := store.ScheduledPost().GetPendingScheduledPosts(beforeTime, afterTime, "", 10) + require.NoError(t, err) + require.Len(t, pending, 1, "pending posts must be read from master, not the empty replica") + require.Equal(t, createdScheduledPost.Id, pending[0].Id) +}