Review changes from upstream #2
Conversation
cmd/wal-listener/init.go
Outdated
| @@ -24,6 +25,12 @@ func initPgxConnections(cfg *config.DatabaseCfg, logger *slog.Logger) (*pgx.Conn | |||
| Password: cfg.Password, | |||
| } | |||
|
|
|||
| if cfg.SSL { | |||
| pgxConf.TLSConfig = &tls.Config{ | |||
| InsecureSkipVerify: true, | |||
There was a problem hiding this comment.
Do we want to be doing this in prod? Can we wrap this in an env check?
listener/listener.go
Outdated
| } | ||
|
|
||
| l.log.Debug("WAL message has been received", slog.Uint64("wal", msg.WalMessage.WalStart)) | ||
| }).Pipe(runtime.NumCPU(), func(raw interface{}) interface{} { |
There was a problem hiding this comment.
Is it the Pipe library that's introducing the use of any and casting between each stage?
Would it not make more sense to use typed channels here?
listener/listener.go
Outdated
| // loop over once triggering the publish of all the events | ||
| for i, event := range batch.events { | ||
| subjectNames[i] = event.SubjectName(l.cfg) | ||
| results[i] = l.publisher.Publish(ctx, subjectNames[i], event) |
There was a problem hiding this comment.
Aren't we going to mess up ordering here?
Events are fully ordered out from the last stage, but now we're fanning out into numCPU amount of goroutines, and wouldn't that allow a batch to be published out of order?
listener/listener.go
Outdated
| // loop over a second time awaiting the result of each publish | ||
| for i, result := range results { | ||
| _, err := result.Get(ctx) | ||
| if err != nil { | ||
| l.monitor.IncProblematicEvents(problemKindPublish) | ||
| return fmt.Errorf("publish: %w", err) |
There was a problem hiding this comment.
This seems kinda problematic, you'll increment a failure event for the first failure but then return out of this whole process. Leaving all the other results just hanging. Potentially missing a lot of problems and having hanging background work.
listener/listener.go
Outdated
|
|
||
| msg, err := l.replicator.WaitForReplicationMessage(ctx) | ||
| if err != nil { | ||
| wg.Add(1) |
There was a problem hiding this comment.
Why do you add to the wg in the failure mode but not when you push a message?
listener/listener.go
Outdated
| l.log.Debug("ack WAL message", slog.Uint64("lsn", l.readLSN())) | ||
| } | ||
| default: | ||
| panic(fmt.Sprintf("unexpected type in pipeline: %T", v)) |
There was a problem hiding this comment.
This feels a bit smelly, if we were using typed channels stuff like this wouldn't be necessary.
listener/pipeline.go
Outdated
|
|
||
| func newPipe(job Job, concurrency int, processErrors bool) *pipe { | ||
| p := &pipe{ | ||
| in: make(jobChan, 1), |
There was a problem hiding this comment.
Why are all of the channels buffered of size 1?
They will block as long as there is a message already in the queue
publisher/pubsub.go
Outdated
| } | ||
| var orderingKey string | ||
| if p.enableOrdering { | ||
| if len(event.PrimaryKey) > 0 { |
There was a problem hiding this comment.
If we have ordering enabled but no primary key this table will be limited to 1Mbps of throughput?
Should we warn or error out in that case?
|
I really think I'd avoid all of the uses of wait group and switch to
|
@airhorns I've been leaving some notes here, but I'm not really sure about some of these changes.