Add stream IDs, proper storage prefixing, recovering from crashes. #177
Conversation
There was a problem hiding this comment.
I think it's high time we started commenting our code a bit more, because just by looking at the code, it is very difficult to understand what's happening in specific parts of the code. I think a two-liner saying "StreamID is ...." and "The transactions are now used in a way X instead of Y" would make reviewing a much more efficient, productive and easy job.
|
|
||
| watermark, err := engine.watermarkSource.GetWatermark(ctx, sourcePrefixedTx) | ||
| watermark, err := engine.watermarkSource.GetWatermark(ctx, tx) |
There was a problem hiding this comment.
This seems like a very subtle change, yet there is nothing that explains it. Could you add a comment, as to explain why the watermark is suddenly created using the original transaction, and not the prefixed one, as was in the previous version?
There was a problem hiding this comment.
I've added necessary comments.
| if err != nil { | ||
| return errors.Wrap(err, "couldn't get current watermark from source") | ||
| } | ||
| if watermark.After(engine.lastCommittedWatermark) { | ||
| err := engine.irs.UpdateWatermark(ctx, tx, watermark) | ||
| err := engine.irs.UpdateWatermark(ctx, prefixedTx, watermark) |
There was a problem hiding this comment.
Same thing here - these changes need commenting in my opinion
There was a problem hiding this comment.
I've added necessary comments.
No description provided.