feat: add rate limiter middleware and update to Go 1.24

feat: add rate limiter middleware and update to Go 1.24

Author: umputun

.github/workflows/ci.yml

@@ -11,10 +11,10 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - name: set up go 1.23
+      - name: set up go 1.24
         uses: actions/setup-go@v3
         with:
-          go-version: "1.23"
+          go-version: "1.24"
         id: go
 
       - name: checkout
@@ -29,9 +29,9 @@ jobs:
           go build -race
 
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v3
+        uses: golangci/golangci-lint-action@v7
         with:
-          version: latest
+          version: v2.1.6
           skip-pkg-cache: true
 
       - name: install goveralls

.golangci.yml

@@ -1,79 +1,83 @@
-linters-settings:
-  govet:
-    shadow: true
-  gocyclo:
-    min-complexity: 15
-  maligned:
-    suggest-new: true
-  goconst:
-    min-len: 2
-    min-occurrences: 2
-  misspell:
-    locale: US
-  lll:
-    line-length: 140
-  gocritic:
-    enabled-tags:
-      - performance
-      - style
-      - experimental
-    disabled-checks:
-      - wrapperFunc
-
+version: "2"
+run:
+  concurrency: 4
 linters:
+  default: none
   enable:
-    - staticcheck
-    - revive
-    - govet
-    - unconvert
-    - gosec
-    - unparam
-    - typecheck
-    - ineffassign
-    - stylecheck
-    - gochecknoinits
-    - copyloopvar
-    - gocritic
-    - nakedret
-    - gosimple
-    - prealloc
-    - unused
     - contextcheck
     - copyloopvar
     - decorder
     - errorlint
     - exptostd
     - gochecknoglobals
-    - gofmt
-    - goimports
+    - gochecknoinits
+    - gocritic
+    - gosec
+    - govet
+    - ineffassign
     - intrange
+    - nakedret
     - nilerr
+    - prealloc
     - predeclared
+    - revive
+    - staticcheck
     - testifylint
     - thelper
-  fast: false
-  disable-all: true
-
-
-run:
-  concurrency: 4
-
-issues:
-  exclude-rules:
-    - text: "G114: Use of net/http serve function that has no support for setting timeouts"
-      linters:
-        - gosec
-    - linters:
-        - unparam
-        - revive
-      path: _test\.go$
-      text: "unused-parameter"
-    - linters:
-        - prealloc
-      path: _test\.go$
-      text: "Consider pre-allocating"
-    - linters:
-        - gosec
-        - intrange
-      path: _test\.go$
-  exclude-use-default: false
+    - unconvert
+    - unparam
+    - unused
+    - nestif
+  settings:
+    goconst:
+      min-len: 2
+      min-occurrences: 2
+    gocritic:
+      disabled-checks:
+        - wrapperFunc
+      enabled-tags:
+        - performance
+        - style
+        - experimental
+    gocyclo:
+      min-complexity: 15
+    govet:
+      enable:
+        - shadow
+    lll:
+      line-length: 140
+    misspell:
+      locale: US
+  exclusions:
+    generated: lax
+    rules:
+      - linters:
+          - gosec
+        text: 'G114: Use of net/http serve function that has no support for setting timeouts'
+      - linters:
+          - revive
+          - unparam
+        path: _test\.go$
+        text: unused-parameter
+      - linters:
+          - prealloc
+        path: _test\.go$
+        text: Consider pre-allocating
+      - linters:
+          - gosec
+          - intrange
+        path: _test\.go$
+    paths:
+      - third_party$
+      - builtin$
+      - examples$
+formatters:
+  enable:
+    - gofmt
+    - goimports
+  exclusions:
+    generated: lax
+    paths:
+      - third_party$
+      - builtin$
+      - examples$

README.md

@@ -225,6 +225,7 @@ The package supports middleware pattern similar to HTTP middleware in Go. Middle
 - Retries with backoff
 - Timeouts
 - Panic recovery
+- Rate limiting
 - Metrics and logging
 - Error handling
 
@@ -242,7 +243,10 @@ p.Use(middleware.Recovery[string](func(p interface{}) {
 }))
 
 // Add validation before processing
-p.Use(middleware.Validate([string]validator))
+p.Use(middleware.Validator[string](validator))
+
+// Add rate limiting
+p.Use(middleware.RateLimiter[string](10, 5))  // 10 requests/sec with burst of 5
 ```
 
 Custom middleware:

examples/collector_errors/go.sum

@@ -0,0 +1,10 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
+golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

examples/collector_errors/main.go

@@ -99,7 +99,7 @@ func main() {
 	fmt.Printf("Processing time:   %v\n", stats.ProcessingTime.Round(time.Millisecond))
 	fmt.Printf("Total time:        %v\n", stats.TotalTime.Round(time.Millisecond))
 
-	// Calculate average duration for successes and failures
+	// calculate average duration for successes and failures
 	var totalSuccessDuration, totalFailureDuration time.Duration
 	for _, s := range successes {
 		totalSuccessDuration += s.Duration
@@ -136,7 +136,7 @@ func main() {
 			results := errorsByType[errType]
 			fmt.Printf("\n• %s (%d occurrences):\n", errType, len(results))
 
-			// Sort results by timestamp
+			// sort results by timestamp
 			sort.Slice(results, func(i, j int) bool {
 				return results[i].Timestamp.Before(results[j].Timestamp)
 			})

examples/direct_chain/main.go

@@ -105,7 +105,7 @@ func ProcessStrings(ctx context.Context, input []string) ([]finalData, error) {
 		results = append(results, v)
 	}
 
-	// Print debug statistics
+	// print debug statistics
 	fmt.Printf("\nProcessing statistics:\n")
 	fmt.Printf("Total items submitted: %d\n", submitted.Load())
 	fmt.Printf("Items passed filter (>2 'a's): %d\n", filtered.Load())

examples/middleware/README.md

@@ -13,6 +13,7 @@ This example demonstrates how to use middleware in [go-pkgz/pool](https://github
     - Input validation before processing
     - Automatic retries for failed tasks
     - Panic recovery for robustness
+    - Rate limiting for flow control
     - Structured logging for observability
 
 3. Real-world patterns:
@@ -26,6 +27,7 @@ This example demonstrates how to use middleware in [go-pkgz/pool](https://github
 - Task validation before processing
 - Automatic retries with exponential backoff
 - Panic recovery with custom handler
+- Rate limiting with token bucket algorithm
 - Structured JSON logging
 - Performance metrics collection
 - Configurable worker count and retry attempts
@@ -72,10 +74,11 @@ The implementation demonstrates several key concepts:
 2. Middleware composition:
    ```go
    pool.New[Task](workers, makeWorker()).Use(
-       middleware.Validate(validator),    // validate first
-       middleware.Retry[Task](retries),  // then retry on failure
+       middleware.Validator(validator),     // validate first
+       middleware.Retry[Task](retries),    // then retry on failure
        middleware.Recovery[Task](handler), // recover from panics
-       customLogger,               // log everything
+       middleware.RateLimiter[Task](5, 3), // rate limit to 5/sec
+       customLogger,                       // log everything
    )
    ```
 
@@ -114,6 +117,20 @@ The implementation demonstrates several key concepts:
     "duration_ms": 100,
     "error": "failed to process task 2"
 }
+{
+    "time": "2025-02-12T10:00:00Z",
+    "level": "INFO",
+    "msg": "submitting rate-limited tasks"
+}
+{
+    "time": "2025-02-12T10:00:00Z",
+    "level": "INFO",
+    "msg": "pool finished",
+    "processed": 14,
+    "errors": 2,
+    "total_time": "3.2s",
+    "duration": "2.1s"
+}
 ```
 
 ## Architecture
@@ -137,4 +154,5 @@ Each component is isolated and has a single responsibility, making the code easy
 - The first middleware wraps the outermost layer
 - Built-in middleware handles common patterns
 - Custom middleware can add any functionality
+- Rate limiting is shared across all workers in the pool
 - Structured logging as an example of cross-cutting concern

examples/middleware/go.mod

@@ -4,6 +4,9 @@ go 1.24
 
 require github.com/go-pkgz/pool v0.7.0
 
-require golang.org/x/sync v0.11.0 // indirect
+require (
+	golang.org/x/sync v0.14.0 // indirect
+	golang.org/x/time v0.11.0 // indirect
+)
 
 replace github.com/go-pkgz/pool => ../..

examples/middleware/go.sum

@@ -4,7 +4,9 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA=
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
-golang.org/x/sync v0.11.0 h1:GGz8+XQP4FvTTrjZPzNKTMFtSXH80RAzG+5ghFPgK9w=
-golang.org/x/sync v0.11.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
+golang.org/x/sync v0.14.0 h1:woo0S4Yywslg6hp4eUFjTVOyKt0RookbpAHG4c1HmhQ=
+golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/time v0.11.0 h1:/bpjEDfN9tkoN/ryeYHnv5hcMlc8ncjMcM4XBk5NWV0=
+golang.org/x/time v0.11.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

examples/middleware/main.go

@@ -79,6 +79,13 @@ func runPool(ctx context.Context, p *pool.WorkerGroup[Task], cfg config) error {
 		p.Submit(task)
 	}
 
+	// demonstrate rate limiting
+	cfg.logger.Info("submitting rate-limited tasks")
+	start := time.Now()
+	for i := 0; i < 10; i++ {
+		p.Submit(Task{ID: fmt.Sprintf("rate-%d", i), Priority: 3, Payload: "rate limited task"})
+	}
+
 	// close pool and wait for completion
 	if err := p.Close(ctx); err != nil {
 		return err
@@ -87,7 +94,7 @@ func runPool(ctx context.Context, p *pool.WorkerGroup[Task], cfg config) error {
 	// print final metrics
 	metrics := p.Metrics().GetStats()
 	cfg.logger.Info("pool finished", "processed", metrics.Processed, "errors", metrics.Errors,
-		"total_time", metrics.TotalTime.String())
+		"total_time", metrics.TotalTime.String(), "duration", time.Since(start).String())
 
 	return nil
 }
@@ -99,7 +106,8 @@ func makePool(cfg config) *pool.WorkerGroup[Task] {
 		middleware.Recovery[Task](func(p interface{}) { // recover from panics
 			cfg.logger.Error("panic recovered", "error", fmt.Sprint(p))
 		}),
-		makeStructuredLogger(cfg.logger), // custom structured logging
+		middleware.RateLimiter[Task](5, 3), // rate limit: 5 tasks/second with burst of 3
+		makeStructuredLogger(cfg.logger),   // custom structured logging
 	)
 }