golibs/go-tasks
Watch
1
0
Fork
You've already forked go-tasks
 0
Code Issues Pull requests Projects Releases 14 Packages Wiki Activity Actions

improvement: Create an ItemSubmitter interface with de-duping and health checks #16

Merged
k-simons merged 21 commits from ksimons/addConsumer into develop 2026-01-05 17:18:21 +00:00
Conversation 4 Commits 21 Files changed 6 +566 -5
k-simons commented 2025-12-19 19:29:29 +00:00 (Migrated from github.com)
  • Create an ItemSubmitter interface with de-duping and health checks
  • Meant for direct application use
  • Updated the Collapsing queue with a few differnet methods
- Create an ItemSubmitter interface with de-duping and health checks - Meant for direct application use - Updated the Collapsing queue with a few differnet methods
changelog-app[bot] commented 2025-12-19 19:29:33 +00:00 (Migrated from github.com)

✅ Successfully generated changelog entry!

Entry generated via PR title

To modify this entry, edit PR title using proper format.

📋Changelog Preview

💡 Improvements

  • Create an ItemSubmitter interface with de-duping and health checks (#16)
### ✅ **Successfully generated changelog entry!** #### Entry generated via PR title To modify this entry, edit PR title using proper format. --- ### 📋Changelog Preview ### 💡 Improvements * Create an ItemSubmitter interface with de-duping and health checks ([#16](https://github.com/palantir/witchcraft-go-tasks/pull/16)) <!-- changelog-success-comment -->
nmiyake (Migrated from github.com) requested changes 2025-12-20 03:13:49 +00:00
nmiyake (Migrated from github.com) left a comment

Thanks! Core logic looks good, but some feedback around parameters and edge cases and a few questions.

Thanks! Core logic looks good, but some feedback around parameters and edge cases and a few questions.
executor/item_submitter.go
@ -0,0 +1,126 @@
// Copyright (c) 2025 Palantir Technologies. All rights reserved.
nmiyake (Migrated from github.com) commented 2025-12-20 02:52:43 +00:00 
	keyedErrorHealthCheckSource window.KeyedErrorHealthCheckSource

Recommend updating name unless there's something particular to k8s about this

```suggestion keyedErrorHealthCheckSource window.KeyedErrorHealthCheckSource ``` Recommend updating name unless there's something particular to k8s about this
nmiyake (Migrated from github.com) commented 2025-12-20 02:53:00 +00:00 
	keyedErrorHealthCheckSource window.KeyedErrorHealthCheckSource,
```suggestion keyedErrorHealthCheckSource window.KeyedErrorHealthCheckSource, ```
nmiyake (Migrated from github.com) commented 2025-12-20 02:59:31 +00:00

Should this be >=?

Should this be >=?
executor/item_submitter.go
@ -0,0 +49,4 @@
// processing happens asynchronously. Duplicate submissions are collapsed.
Submit(context.Context, T)
}
type defaultItemSubmitter[T constraint] struct {
nmiyake (Migrated from github.com) commented 2025-12-20 02:52:12 +00:00 
}

type defaultItemSubmitter[T constraint] struct {
```suggestion } type defaultItemSubmitter[T constraint] struct { ```
executor/item_submitter.go
@ -0,0 +89,4 @@
for {
element, shutdown := d.queue.Get()
if shutdown {
svc1log.FromContext(ctx).Warn("Queue shutting down; workers stopping.")
nmiyake (Migrated from github.com) commented 2025-12-20 02:54:27 +00:00

Is there a particular reason that this is Warn rather than Info? Just wondering, as it seems like this behavior might be expected in the normal operation.

Is there a particular reason that this is `Warn` rather than `Info`? Just wondering, as it seems like this behavior might be expected in the normal operation.
executor/item_submitter.go
@ -0,0 +117,4 @@
func (d defaultItemSubmitter[T]) handleProcessError(ctx context.Context, element T, err error) {
numRequeues := d.queue.NumRequeues(element)
ctx = svc1log.WithLoggerParams(ctx, svc1log.SafeParam("maxNumRequeues", d.maxNumRequeues), svc1log.SafeParam("numRequeues", numRequeues))
d.logError(ctx, err)
nmiyake (Migrated from github.com) commented 2025-12-20 02:59:07 +00:00

Should either check d.logError != nil here or update param to ensure that a nil value cannot be set

Should either check `d.logError != nil` here or update param to ensure that a `nil` value cannot be set
executor/item_submitter_option.go
@ -0,0 +35,4 @@
// Any value under 1 will cause 0 requeues to occur
func WithMaxNumRequeues[T constraint](maxNumRequeues int) ItemSubmitterOption[T] {
return itemSubmitterOptionFunc[T](func(defaultItemSubmitterArg *defaultItemSubmitter[T]) {
defaultItemSubmitterArg.maxNumRequeues = maxNumRequeues
nmiyake (Migrated from github.com) commented 2025-12-20 02:57:51 +00:00

What is the behavior if the provided value is <=0?

What is the behavior if the provided value is <=0?
executor/item_submitter_option.go
@ -0,0 +46,4 @@
if errorLogger == nil {
return
}
defaultItemSubmitterArg.logError = errorLogger
nmiyake (Migrated from github.com) commented 2025-12-20 03:01:55 +00:00

Per other comment, need to either add a check here to ensure that errorLogger is not nil or add a check when errorLogger is invoked to ensure that it is not invoked if it is nil.

Per other comment, need to either add a check here to ensure that `errorLogger` is not `nil` or add a check when `errorLogger` is invoked to ensure that it is not invoked if it is `nil`.
internal/queue/collapsing_queue.go
@ -37,16 +38,28 @@ type CollapsingQueue[T comparable] interface {
// queue while it was being processed, it will be re-queued for another processing
nmiyake (Migrated from github.com) commented 2025-12-20 03:04:09 +00:00

Is Forget the standard nomenclature? ResetRateLimit seems more natural/understandable to me, but don't feel strongly about it so fine to keep the naming as well (just a question/suggestion).

Is `Forget` the standard nomenclature? `ResetRateLimit` seems more natural/understandable to me, but don't feel strongly about it so fine to keep the naming as well (just a question/suggestion).
internal/queue/rate_limiter.go
@ -0,0 +1,80 @@
// Copyright (c) 2025 Palantir Technologies. All rights reserved.
nmiyake (Migrated from github.com) commented 2025-12-20 03:10:04 +00:00 
// ItemExponentialFailureRateLimiter does a simple baseDelay*2^<num-failures> limit
```suggestion // ItemExponentialFailureRateLimiter does a simple baseDelay*2^<num-failures> limit ```
nmiyake (Migrated from github.com) commented 2025-12-20 03:12:52 +00:00

What is considered a "failure" here? It's not really clear from looking at the interface documentation. Looking at the implementation, it looks like it might be the case that any call to When increments the "failure" count? It's not really clear to me why that would be the case, though.

What is considered a "failure" here? It's not really clear from looking at the interface documentation. Looking at the implementation, it looks like it might be the case that any call to `When` increments the "failure" count? It's not really clear to me why that would be the case, though.
internal/queue/rate_limiter.go
@ -0,0 +64,4 @@
if calculated > r.maxDelay {
return r.maxDelay
}
return calculated
nmiyake (Migrated from github.com) commented 2025-12-20 03:09:33 +00:00

Fine to leave as-is as well, but min builtin can also be used:

    return min(time.Duration(backoff), r.maxDelay)
Fine to leave as-is as well, but `min` builtin can also be used: ```suggestion return min(time.Duration(backoff), r.maxDelay) ```
nmiyake (Migrated from github.com) approved these changes 2026-01-05 17:17:04 +00:00
autorelease3[bot] commented 2026-01-05 17:18:29 +00:00 (Migrated from github.com)

Released v0.6.0

Released [v0.6.0](https://github.com/palantir/witchcraft-go-tasks/releases/tag/v0.6.0)
Sign in to join this conversation.
Reviewers
No reviewers
nmiyake
Labels
Clear labels   
🤖 fix nits

   
🧠 ai fix nits

   
autorelease

   
bug
Something isn't working

  
do not merge

   
documentation
Improvements or additions to documentation

  
duplicate
This issue or pull request already exists

  
enhancement
New feature or request

  
good first issue
Good for newcomers

  
help wanted
Extra attention is needed

  
invalid
This doesn't seem right

  
merge when ready

   
no changelog

   
question
Further information is requested

  
wontfix
This will not be worked on

No labels
🤖 fix nits
🧠 ai fix nits
autorelease
bug
do not merge
documentation
duplicate
enhancement
good first issue
help wanted
invalid
merge when ready
no changelog
question
wontfix
Milestone
Clear milestone
No items
No milestone
Projects
Clear projects
No items
No project
Assignees
Clear assignees
No assignees
1 participant
Notifications
Due date
The due date is invalid or out of range. Please use the format "yyyy-mm-dd".

No due date set.

Dependencies

No dependencies set

Reference
golibs/go-tasks!16
No description provided.