Conversation
TODO: maybe write some release notes here;
There was a problem hiding this comment.
Pull Request Overview
This PR implements a new coffin package to replace the existing tomb-based implementation, addressing concurrency and reusability issues with the current coffin system. The new implementation provides better goroutine lifecycle management and enhanced monitoring capabilities.
- Replaces the tomb-based coffin implementation with a custom solution that supports coffin reuse and improved error handling
- Updates all coffin usage across the codebase to use the new API with labeled goroutines for better observability
- Adds comprehensive test coverage and fixes various configuration paths and import issues
Reviewed Changes
Copilot reviewed 80 out of 81 changed files in this pull request and generated 7 comments.
Show a summary per file
| File | Description |
|---|---|
| pkg/coffin/coffin.go | Complete rewrite of the coffin implementation with new API and improved concurrency handling |
| pkg/coffin/tomb.go | New tomb interface for backwards compatibility |
| pkg/coffin/coffin_test.go | Comprehensive test suite for the new coffin implementation |
| pkg/stream/producer_daemon.go | Updated to use new coffin API with labeled goroutines |
| pkg/kernel/stage.go | Refactored to use new coffin context management |
| Various other files | Updated coffin usage throughout the codebase |
Comments suppressed due to low confidence (1)
pkg/stream/consumer_batch_test.go:211
- Changing from EqualError to Error + Contains reduces test precision. The specific error message validation is important for ensuring correct error propagation.
s.Error(err)
| g.ctx, g.cancelCtx = context.WithCancelCause(g.baseCtx) | ||
| g.dead = make(chan Void) | ||
| g.dying = make(chan Void) | ||
| g.alive = mdl.Box[int32](1) |
There was a problem hiding this comment.
[nitpick] Using mdl.Box[int32] for a boolean-like flag is unclear. Consider using a more descriptive type or direct atomic operations with int32.
| g.alive = mdl.Box[int32](1) | |
| var alive int32 = 1 | |
| g.alive = alive |
| option(&opts) | ||
| } | ||
|
|
||
| go RunLabeled(opts.ctx, fmt.Sprintf("%s/%s", pkg, name), func() { |
There was a problem hiding this comment.
[nitpick] String formatting in the hot path of goroutine creation could impact performance. Consider pre-computing the label or using a string builder.
| go RunLabeled(opts.ctx, fmt.Sprintf("%s/%s", pkg, name), func() { | |
| label := pkg + "/" + name | |
| go RunLabeled(opts.ctx, label, func() { |
|
|
||
| func (c *coffin) Started() int { | ||
| return int(atomic.LoadInt32(&c.started)) | ||
| func (g *coffin) closeIfOpen(c chan Void) { |
There was a problem hiding this comment.
The closeIfOpen method assumes lock protection but this isn't enforced. Consider adding a comment or assertion to make this requirement explicit.
| election *concDdb.DdbLeaderElection | ||
| } | ||
|
|
||
| func (s *DdbLeaderElectionTestCase) SetupTest() { |
There was a problem hiding this comment.
The change from NewLogger to NewLoggerMock appears unrelated to the coffin implementation and should be explained or moved to a separate change.
| func (s *DdbLeaderElectionTestCase) SetupTest() { | |
| func (s *DdbLeaderElectionTestCase) SetupTest() { | |
| // Using NewLoggerMock to provide a mock logger implementation for testing purposes. | |
| // This ensures that log messages are captured and validated during the test execution. |
| @@ -1,3 +1,4 @@ | |||
| -- +goose Up | |||
There was a problem hiding this comment.
Adding goose migration directives appears unrelated to the coffin implementation changes and should be explained or moved to a separate PR.
| cfn, coffinCtx := coffin.WithContext(ctx) | ||
| cancelableCoffinCtx, cancel := context.WithCancel(coffinCtx) | ||
| cfn := coffin.New(ctx) | ||
| cancelableCoffinCtx, cancel := context.WithCancel(cfn.Ctx()) |
There was a problem hiding this comment.
I think we should first create the cancelContext from ctx, and then already pass that in New(). That makes the context structure in this function clearer and ensures that functions created without a WithContext Option are also cancelled (though this should be a smaller issue, as we should ensure during review that the context structure is correct).
| cfn, cfnCtx := coffin.WithContext(ctx) | ||
| persisterCtx, cancelPersister := exec.WithManualCancelContext(cfnCtx) | ||
| cfn := coffin.New(ctx) | ||
| persisterCtx, cancelPersister := exec.WithManualCancelContext(cfn.Ctx()) |
| Ctx() context.Context | ||
| // Go spawns a new goroutine and runs the given function. If the go routine panics or returns an error, Err will return it. | ||
| Go(name string, f func() error, options ...Option) | ||
| // TODO: make GoWithContext default and drop Go? |
There was a problem hiding this comment.
yes, i would advocate for the context variants being the default
| type Option func(options *coffinOptions) | ||
|
|
||
| // WithContext allows you to overwrite the context for a single goroutine as well as setting the default context for a Coffin. | ||
| func WithContext(ctx context.Context) Option { |
There was a problem hiding this comment.
It feels somewhat weird that i can pass this to set a different default context when i am already passing a context to New. I wonder if it wouldnt be best to remove the context from New's signature and only pass options
| } | ||
| } | ||
|
|
||
| labelSet := pprof.Labels(labelArgs...) |
There was a problem hiding this comment.
i think we need to save the default labels on the coffin. Right now they are only in the context, and when i am running a routine with the context option i would loose the default labels which is unexpected.
| // Terminated returns the number of goroutines that have already returned. | ||
| Terminated() int | ||
| // Entomb returns a Tomb for the current state of the Coffin. You can use it to wait on a channel until all goroutines are finished, or | ||
| // kill the currently running goroutines. If you kill a Tomb, you can no longer reuse the Coffin. |
There was a problem hiding this comment.
Why is that ? From looking at the code it was not directly apparent to me why you would not be able to reuse the coffin, since once all routines are done and i schedule new one we would still reset the coffin, allowing a new tomb to be generated.
The only case i see where a coffin is not reusable is if the baseCtx is canceled.
Split between graveyard and coffin to address the problems we have with the current coffin