Skip to content

Commit ad28fad

Browse files
atzoumatzoum
committed
feat(processor): fork event processing for configured isolated destinations
chore: fixes chore: partition buffer store consistency chore: improvements
1 parent f471b1d commit ad28fad

47 files changed

Lines changed: 1277 additions & 106 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎app/apphandlers/apphandlers_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -226,7 +226,7 @@ func setDefaultEnv(t *testing.T) {
226226
t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "SourceDebugger.disableEventUploads"), "true")
227227
t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "TransformationDebugger.disableTransformationStatusUploads"), "true")
228228
t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "JobsDB.backup.enabled"), "false")
229-
t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "JobsDB.migrateDSLoopSleepDuration"), "60m")
229+
t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "JobsDB.compactionLoopSleepDuration"), "60m")
230230
t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "archival.Enabled"), "false")
231231
t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "Reporting.syncer.enabled"), "false")
232232
t.Setenv(config.ConfigKeyToEnv(config.DefaultEnvPrefix, "BatchRouter.mainLoopFreq"), "1s")

‎app/apphandlers/embeddedAppHandler.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -208,6 +208,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
208208
gwWOHandle := jobsdb.NewForWrite(
209209
"gw",
210210
jobsdb.WithClearDB(options.ClearDB),
211+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw jobsdb
211212
jobsdb.WithStats(statsFactory),
212213
jobsdb.WithDBHandle(jobsdbPool),
213214
jobsdb.WithNumPartitions(partitionCount),
@@ -223,6 +224,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
223224

224225
gwROHandle := jobsdb.NewForRead(
225226
"gw",
227+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw jobsdb
226228
jobsdb.WithDSLimit(a.config.gwDSLimit),
227229
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "Gateway.jobsDB.skipMaintenanceError")),
228230
jobsdb.WithStats(statsFactory),
@@ -265,6 +267,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
265267
eschRWDB := jobsdb.NewForReadWrite(
266268
"esch",
267269
jobsdb.WithClearDB(options.ClearDB),
270+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in esch jobsdb
268271
jobsdb.WithDSLimit(a.config.eschDSLimit),
269272
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(false, "Processor.jobsDB.skipMaintenanceError")),
270273
jobsdb.WithStats(statsFactory),
@@ -276,6 +279,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
276279
arcRWDB := jobsdb.NewForReadWrite(
277280
"arc",
278281
jobsdb.WithClearDB(options.ClearDB),
282+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in arc jobsdb
279283
jobsdb.WithDSLimit(a.config.arcDSLimit),
280284
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(false, "Processor.jobsDB.skipMaintenanceError")),
281285
jobsdb.WithStats(statsFactory),
@@ -291,6 +295,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, shutdownFn func(), op
291295
"proc",
292296
jobsdb.WithMultiConsumer(),
293297
jobsdb.WithClearDB(options.ClearDB),
298+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in proc jobsdb
294299
jobsdb.WithDSLimit(a.config.procDSLimit),
295300
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(false, "Processor.jobsDB.skipMaintenanceError")),
296301
jobsdb.WithStats(statsFactory),

‎app/apphandlers/gatewayAppHandler.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,7 @@ func (a *gatewayApp) StartRudderCore(ctx context.Context, _ func(), options *app
104104
gwWOHandle := jobsdb.NewForWrite(
105105
"gw",
106106
jobsdb.WithClearDB(options.ClearDB),
107+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw jobsdb
107108
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "Gateway.jobsDB.skipMaintenanceError")),
108109
jobsdb.WithStats(statsFactory),
109110
jobsdb.WithDBHandle(jobsdbPool),

‎app/apphandlers/processorAppHandler.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -210,6 +210,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
210210

211211
gwROHandle := jobsdb.NewForRead(
212212
"gw",
213+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw jobsdb
213214
jobsdb.WithDSLimit(a.config.gwDSLimit),
214215
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "Gateway.jobsDB.skipMaintenanceError")),
215216
jobsdb.WithStats(statsFactory),
@@ -250,6 +251,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
250251
eschRWDB := jobsdb.NewForReadWrite(
251252
"esch",
252253
jobsdb.WithClearDB(options.ClearDB),
254+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in esch jobsdb
253255
jobsdb.WithDSLimit(a.config.eschDSLimit),
254256
jobsdb.WithStats(statsFactory),
255257
jobsdb.WithDBHandle(jobsdbPool),
@@ -260,6 +262,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
260262
arcRWDB := jobsdb.NewForReadWrite(
261263
"arc",
262264
jobsdb.WithClearDB(options.ClearDB),
265+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in arc jobsdb
263266
jobsdb.WithDSLimit(a.config.arcDSLimit),
264267
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(false, "Processor.jobsDB.skipMaintenanceError")),
265268
jobsdb.WithStats(statsFactory),
@@ -275,6 +278,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, shutdownFn func(), o
275278
"proc",
276279
jobsdb.WithMultiConsumer(),
277280
jobsdb.WithClearDB(options.ClearDB),
281+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in proc jobsdb
278282
jobsdb.WithDSLimit(a.config.procDSLimit),
279283
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(false, "Processor.jobsDB.skipMaintenanceError")),
280284
jobsdb.WithStats(statsFactory),

‎app/apphandlers/setup_partitionmigration.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
9595
gwBuffRWHandle := jobsdb.NewForReadWrite(
9696
"gw_buf",
9797
jobsdb.WithClearDB(false),
98+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw_buf jobsdb
9899
jobsdb.WithDSLimit(config.GetReloadableIntVar(0, 1, "JobsDB.gw_buf.dsLimit", "JobsDB.dsLimit")),
99100
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "JobsDB.gw_buf.skipMaintenanceError", "JobsDB.buff.skipMaintenanceError", "JobsDB.skipMaintenanceError")),
100101
jobsdb.WithStats(stats),
@@ -115,6 +116,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
115116
gwWODB = jobsdb.NewForWrite(
116117
"gw",
117118
jobsdb.WithClearDB(false),
119+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw jobsdb
118120
jobsdb.WithStats(stats),
119121
jobsdb.WithDBHandle(dbPool),
120122
jobsdb.WithPriorityPoolDB(priorityPool),
@@ -123,6 +125,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
123125
)
124126
gwBuffROHandle := jobsdb.NewForRead(
125127
"gw_buf",
128+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw_buf jobsdb
126129
jobsdb.WithDSLimit(config.GetReloadableIntVar(0, 1, "JobsDB.gw_buf.dsLimit", "JobsDB.dsLimit")),
127130
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "JobsDB.gw_buf.skipMaintenanceError", "JobsDB.buff.skipMaintenanceError", "JobsDB.skipMaintenanceError")),
128131
jobsdb.WithStats(stats),
@@ -153,6 +156,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
153156
rtBuffRWHandle := jobsdb.NewForReadWrite(
154157
"rt_buf",
155158
jobsdb.WithClearDB(false),
159+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in rt_buf jobsdb
156160
jobsdb.WithDSLimit(config.GetReloadableIntVar(0, 1, "JobsDB.rt_buff.dsLimit", "JobsDB.dsLimit")),
157161
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "JobsDB.rt_buff.skipMaintenanceError", "JobsDB.buff.skipMaintenanceError", "JobsDB.skipMaintenanceError")),
158162
jobsdb.WithStats(stats),
@@ -180,6 +184,7 @@ func setupProcessorPartitionMigrator(ctx context.Context,
180184
brtBuffRWHandle := jobsdb.NewForReadWrite(
181185
"batch_rt_buf",
182186
jobsdb.WithClearDB(false),
187+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in batch_rt_buf jobsdb
183188
jobsdb.WithDSLimit(config.GetReloadableIntVar(0, 1, "JobsDB.batch_rt_buff.dsLimit", "JobsDB.dsLimit")),
184189
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "JobsDB.batch_rt_buff.skipMaintenanceError", "JobsDB.buff.skipMaintenanceError", "JobsDB.skipMaintenanceError")),
185190
jobsdb.WithStats(stats),
@@ -208,7 +213,10 @@ func setupProcessorPartitionMigrator(ctx context.Context,
208213
if procRWDB != nil {
209214
procBuffRWHandle := jobsdb.NewForReadWrite(
210215
"proc_buf",
216+
// proc_buf doesn't need to be multi-consumer because it is used only as a temporary storage.
217+
// During flush, a single consumer will move the jobs to the proc jobsdb which is multi-consumer.
211218
jobsdb.WithClearDB(false),
219+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in proc_buf jobsdb
212220
jobsdb.WithDSLimit(config.GetReloadableIntVar(0, 1, "JobsDB.proc_buf.dsLimit", "JobsDB.dsLimit")),
213221
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "JobsDB.proc_buf.skipMaintenanceError", "JobsDB.buff.skipMaintenanceError", "JobsDB.skipMaintenanceError")),
214222
jobsdb.WithStats(stats),
@@ -347,6 +355,7 @@ func setupGatewayPartitionMigrator(ctx context.Context,
347355
gwBuffWOHandle := jobsdb.NewForWrite(
348356
"gw_buf",
349357
jobsdb.WithClearDB(false),
358+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in gw_buf jobsdb
350359
jobsdb.WithSkipMaintenanceErr(config.GetBoolVar(true, "JobsDB.gw_buf.skipMaintenanceError", "JobsDB.buff.skipMaintenanceError", "JobsDB.skipMaintenanceError")),
351360
jobsdb.WithStats(stats),
352361
jobsdb.WithDBHandle(dbPool),

‎archiver/archiver_isolation_test.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -160,7 +160,7 @@ func ArchivalScenario(
160160
config.Set("SourceDebugger.disableEventUploads", true)
161161
config.Set("TransformationDebugger.disableTransformationStatusUploads", true)
162162
config.Set("JobsDB.backup.enabled", false)
163-
config.Set("JobsDB.migrateDSLoopSleepDuration", "60m")
163+
config.Set("JobsDB.compactionLoopSleepDuration", "60m")
164164
config.Set("JobsDB.enableWriterQueue", false)
165165
config.Set("RUDDER_TMPDIR", os.TempDir())
166166
config.Set("archival.ArchiveSleepDuration", "1s")
@@ -322,7 +322,9 @@ func insertJobs(
322322
configMap map[string]backendconfig.ConfigT,
323323
numJobsPerSource int,
324324
) (map[string][]*jobsdb.JobT, int) {
325-
gwJobsDB := jobsdb.NewForWrite("gw", jobsdb.WithStats(stats.NOP))
325+
gwJobsDB := jobsdb.NewForWrite("gw", jobsdb.WithStats(stats.NOP),
326+
jobsdb.WithDefaultSkipStatusCompaction(true), // must match the default used by every other "gw" jobsdb construction
327+
)
326328
require.NoError(t, gwJobsDB.Start(), "it should be able to start the jobsdb")
327329
defer gwJobsDB.Stop()
328330

‎cluster/migrator/processor/targetnode/targetnode_migrator_builder_test.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -230,3 +230,7 @@ func (m *mockBufferedJobsDB) RefreshBufferedPartitions(ctx context.Context) erro
230230
func (m *mockBufferedJobsDB) FlushBufferedPartitions(ctx context.Context, partitionIds []string) error {
231231
return nil
232232
}
233+
234+
func (m *mockBufferedJobsDB) WithStoreConsistency(ctx context.Context, fn func() error) error {
235+
return fn()
236+
}

‎cluster/partitionbuffer/jobsdb_partition_buffer.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,10 @@ type JobsDBPartitionBuffer interface {
3131
RefreshBufferedPartitions(ctx context.Context) error
3232
// FlushBufferedPartitions flushes the buffered data for the provided partition ids to the database and unmarks them as buffered.
3333
FlushBufferedPartitions(ctx context.Context, partitionIds []string) error
34+
// WithStoreConsistency runs fn while keeping Store routing consistent with concurrent flush switchovers.
35+
// Callers storing into an externally-owned transaction (via WithStoreSafeTxFromTx + StoreInTx) must wrap
36+
// that whole transaction with it.
37+
WithStoreConsistency(ctx context.Context, fn func() error) error
3438
}
3539

3640
type jobsDBPartitionBuffer struct {

‎cluster/partitionbuffer/jobsdb_partition_buffer_store.go‎

Lines changed: 31 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -27,23 +27,21 @@ func (b *jobsDBPartitionBuffer) WithStoreSafeTx(ctx context.Context, fn func(tx
2727
return ErrStoreNotSupported
2828
}
2929
for {
30-
if !b.bufferedPartitionsMu.RTryLockWithContext(ctx) {
31-
return fmt.Errorf("acquiring a buffered partitions read lock: %w", ctx.Err())
32-
}
33-
err := b.primaryWriteJobsDB.WithStoreSafeTx(ctx, func(tx jobsdb.StoreSafeTx) (err error) {
34-
if !b.differentBufferDBs { // no need to check for stale version
30+
err := b.WithStoreConsistency(ctx, func() error {
31+
return b.primaryWriteJobsDB.WithStoreSafeTx(ctx, func(tx jobsdb.StoreSafeTx) (err error) {
32+
if !b.differentBufferDBs { // no need to check for stale version
33+
return fn(tx)
34+
}
35+
diff, err := b.versionDiff(ctx, tx.Tx()) // get the version difference
36+
if err != nil {
37+
return err
38+
}
39+
if diff != 0 { // stale version
40+
return errStaleBufferedPartitions
41+
}
3542
return fn(tx)
36-
}
37-
diff, err := b.versionDiff(ctx, tx.Tx()) // get the version difference
38-
if err != nil {
39-
return err
40-
}
41-
if diff != 0 { // stale version
42-
return errStaleBufferedPartitions
43-
}
44-
return fn(tx)
43+
})
4544
})
46-
b.bufferedPartitionsMu.RUnlock()
4745
if !errors.Is(err, errStaleBufferedPartitions) {
4846
return err
4947
}
@@ -54,6 +52,24 @@ func (b *jobsDBPartitionBuffer) WithStoreSafeTx(ctx context.Context, fn func(tx
5452
}
5553
}
5654

55+
// WithStoreConsistency runs fn while holding the buffered-partitions read lock, so that Store
56+
// routing decisions (see splitJobs) stay consistent with a concurrent flush switchover, which
57+
// drains and unbuffers partitions under the corresponding write lock.
58+
//
59+
// The lock must be acquired before the transaction fn stores into is opened and held until that
60+
// transaction commits: a buffered write only becomes visible to a switchover's drain (a separate
61+
// connection) once it commits, so releasing the lock any earlier lets a switchover unbuffer the
62+
// partition in between and orphan the job. Store/WithStoreSafeTx use this internally; callers that
63+
// store into an externally-owned transaction (via WithStoreSafeTxFromTx + StoreInTx) must wrap that
64+
// whole transaction with it themselves.
65+
func (b *jobsDBPartitionBuffer) WithStoreConsistency(ctx context.Context, fn func() error) error {
66+
if !b.bufferedPartitionsMu.RTryLockWithContext(ctx) {
67+
return fmt.Errorf("acquiring a buffered partitions read lock: %w", ctx.Err())
68+
}
69+
defer b.bufferedPartitionsMu.RUnlock()
70+
return fn()
71+
}
72+
5773
// StoreInTx stores the provided jobs into the appropriate JobsDBs based on their partition buffering status within the provided StoreSafeTx
5874
func (b *jobsDBPartitionBuffer) StoreInTx(ctx context.Context, tx jobsdb.StoreSafeTx, jobList []*jobsdb.JobT) error {
5975
if !b.canStore {

‎enterprise/reporting/error_index/error_index_reporting.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -201,6 +201,7 @@ func (eir *ErrorIndexReporter) DatabaseSyncer(c types.SyncerConfig) types.Report
201201
errIndexDB := jobsdb.NewForReadWrite(
202202
"err_idx",
203203
jobsdb.WithDBHandle(dbHandle),
204+
jobsdb.WithDefaultSkipStatusCompaction(true), // no failed job statuses in err_idx jobsdb
204205
jobsdb.WithDSLimit(eir.conf.GetReloadableIntVar(0, 1, "Reporting.errorIndexReporting.dsLimit")),
205206
jobsdb.WithConfig(eir.conf),
206207
jobsdb.WithSkipMaintenanceErr(eir.conf.GetBoolVar(false, "Reporting.errorIndexReporting.skipMaintenanceError")),

0 commit comments

Comments
 (0)