From 0675207a312e3ef8ee60c99b930ae0c59554a659 Mon Sep 17 00:00:00 2001
From: Viacheslav Poturaev <vearutop@gmail.com>
Date: Tue, 4 Aug 2026 10:47:48 +0200
Subject: [PATCH 1/2] Add dynamic rate limiting and HTTP control interface

---
 .github/workflows/golangci-lint.yml |  23 +---
 .golangci.yml                       |  17 +--
 Makefile                            |   2 +-
 cmd/catp/README.md                  |  21 ++++
 cmd/catp/catp/app.go                |  25 ++++
 cmd/catp/catp/catp.go               | 173 +++++++++++++++++++++++++---
 go.mod                              |   2 +-
 go.sum                              |   4 +-
 8 files changed, 213 insertions(+), 54 deletions(-)

diff --git a/.github/workflows/golangci-lint.yml b/.github/workflows/golangci-lint.yml
index d9e1bb6..34c9ed4 100644
--- a/.github/workflows/golangci-lint.yml
+++ b/.github/workflows/golangci-lint.yml
@@ -24,25 +24,6 @@ jobs:
           go-version: stable
       - uses: actions/checkout@v4
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v8.0.0
+        uses: golangci/golangci-lint-action@v9.2.0
         with:
-          # Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version.
-          version: v2.5.0
-
-          # Optional: working directory, useful for monorepos
-          # working-directory: somedir
-
-          # Optional: golangci-lint command line arguments.
-          # args: --issues-exit-code=0
-
-          # Optional: show only new issues if it's a pull request. The default value is `false`.
-          # only-new-issues: true
-
-          # Optional: if set to true then the action will use pre-installed Go.
-          # skip-go-installation: true
-
-          # Optional: if set to true then the action don't cache or restore ~/go/pkg.
-          # skip-pkg-cache: true
-
-          # Optional: if set to true then the action don't cache or restore ~/.cache/go-build.
-          # skip-build-cache: true
\ No newline at end of file
+          version: v2.11.3
diff --git a/.golangci.yml b/.golangci.yml
index a25123a..0a528da 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -5,12 +5,8 @@ run:
 linters:
   default: all
   disable:
-    - funlen
-    - gocyclo
-    - gocognit
-    - err113
     - embeddedstructfieldcheck
-    - testpackage
+    - nilnil
     - noinlineerr
     - wsl_v5
     - funcorder
@@ -36,22 +32,13 @@ linters:
     - varnamelen
     - wrapcheck
   settings:
-    funlen:
-      lines: 150
-      statements: 50
-    cyclop:
-      max-complexity: 35
-    gocognit:
-      min-complexity: 50
-    nestif:
-      min-complexity: 12
     dupl:
       threshold: 100
     errcheck:
       check-type-assertions: true
       check-blank: true
     gocyclo:
-      min-complexity: 25
+      min-complexity: 20
     misspell:
       locale: US
     unparam:
diff --git a/Makefile b/Makefile
index 1e11310..c366297 100644
--- a/Makefile
+++ b/Makefile
@@ -1,4 +1,4 @@
-#GOLANGCI_LINT_VERSION := "v2.5.0" # Optional configuration to pinpoint golangci-lint version.
+#GOLANGCI_LINT_VERSION := "v2.11.3" # Optional configuration to pinpoint golangci-lint version.
 
 # The head of Makefile determines location of dev-go to include standard targets.
 GO ?= go
diff --git a/cmd/catp/README.md b/cmd/catp/README.md
index 2749360..3275429 100644
--- a/cmd/catp/README.md
+++ b/cmd/catp/README.md
@@ -32,6 +32,12 @@ catp [OPTIONS] PATH ...
   -end-line int
         stop printing lines at this line (exclusive),
         default is 0 (no limit), each input file is counted separately
+  -http string
+        address to serve rate limit control interface on (e.g. :6060),
+        / shows a status page, GET /rate-limit reads the current limit,
+        POST /rate-limit?value=X sets a new one (0 or less means unlimited)
+  -http-auth string
+        user:pass to require as HTTP basic auth for -http endpoints
   -l    count lines
   -no-progress
         disable progress printing
@@ -135,6 +141,21 @@ Each source line would follow the filtering pipeline:
   skipped
 * if not, source line gets into output because of `-pass-any`
 
+### Control rate limit at runtime over HTTP
+
+Start unlimited (or with an initial `-rate-limit`), then throttle or resume from another terminal.
+
+```
+catp -http :6060 -http-auth admin:secret -rate-limit 1000 -output out.log input.log
+```
+
+```
+curl -u admin:secret http://localhost:6060/               # status page
+curl -u admin:secret http://localhost:6060/rate-limit      # read current limit
+curl -u admin:secret -X POST "http://localhost:6060/rate-limit?value=100"  # throttle down
+curl -u admin:secret -X POST "http://localhost:6060/rate-limit?value=0"    # unlimited
+```
+
 ### Split matches into separate files
 
 ```
diff --git a/cmd/catp/catp/app.go b/cmd/catp/catp/app.go
index 559ebd2..5258f2c 100644
--- a/cmd/catp/catp/app.go
+++ b/cmd/catp/catp/app.go
@@ -4,6 +4,7 @@ package catp
 import (
 	"bufio"
 	"bytes"
+	"context"
 	"errors"
 	"flag"
 	"fmt"
@@ -20,6 +21,7 @@ import (
 
 	"github.com/bool64/dev/version"
 	"github.com/bool64/progress"
+	"golang.org/x/time/rate"
 )
 
 // Main is the entry point for catp CLI tool.
@@ -90,6 +92,10 @@ func Main(options ...func(o *Options)) error { //nolint:funlen,cyclop,gocognit,g
 	flag.BoolVar(&r.noProgress, "no-progress", false, "disable progress printing")
 	flag.BoolVar(&r.countLines, "l", false, "count lines")
 	flag.Float64Var(&r.rateLimit, "rate-limit", 0, "output rate limit lines per second")
+	flag.StringVar(&r.httpAddr, "http", "", "address to serve rate limit control interface on (e.g. :6060),\n"+
+		"/ shows a status page, GET /rate-limit reads the current limit,\n"+
+		"POST /rate-limit?value=X sets a new one (0 or less means unlimited)")
+	flag.StringVar(&r.httpAuth, "http-auth", "", "user:pass to require as HTTP basic auth for -http endpoints")
 	progressJSON := flag.String("progress-json", "", "write current progress to a file")
 	ver := flag.Bool("version", false, "print version and exit")
 
@@ -235,6 +241,24 @@ func Main(options ...func(o *Options)) error { //nolint:funlen,cyclop,gocognit,g
 		r.sizes[fn] = st.Size()
 	}
 
+	if r.rateLimit > 0 || r.httpAddr != "" {
+		lim := rate.Inf
+		if r.rateLimit > 0 {
+			lim = rate.Limit(r.rateLimit)
+		}
+
+		r.limiter = rate.NewLimiter(lim, 100)
+	}
+
+	if r.httpAddr != "" {
+		if err := r.startControlServer(); err != nil {
+			return err
+		}
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	r.ctx = ctx
+
 	shutdown := make(chan os.Signal, 1)
 	signal.Notify(shutdown, syscall.SIGINT, syscall.SIGTERM)
 
@@ -243,6 +267,7 @@ func Main(options ...func(o *Options)) error { //nolint:funlen,cyclop,gocognit,g
 
 		println("received signal, shutting down...")
 		atomic.StoreInt64(&r.closed, 1)
+		cancel()
 	}()
 
 	if r.parallel >= 2 {
diff --git a/cmd/catp/catp/catp.go b/cmd/catp/catp/catp.go
index 2e85545..e594ed1 100644
--- a/cmd/catp/catp/catp.go
+++ b/cmd/catp/catp/catp.go
@@ -3,14 +3,18 @@ package catp
 import (
 	"bufio"
 	"context"
+	"crypto/subtle"
 	"encoding/csv"
 	"encoding/json"
 	"fmt"
 	"io"
 	"log"
+	"net"
+	"net/http"
 	"os"
 	"path"
 	"runtime/pprof"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -55,6 +59,16 @@ type runner struct {
 
 	rateLimit float64
 	limiter   *rate.Limiter
+	httpAddr  string
+	httpAuth  string
+	ctx       context.Context
+
+	skipPos      int64
+	lastSkipPos  int64
+	lastSkipTime int64
+
+	statusMu   sync.Mutex
+	lastStatus string
 
 	noProgress bool
 	countLines bool
@@ -168,6 +182,29 @@ func (r *runner) st(s progress.Status) string {
 		atomic.StoreInt64(&r.lastBytesUncompressed, currentBytesUncompressed)
 	}
 
+	if r.parallel <= 1 && r.startLine > 0 {
+		target := int64(r.startLine)
+
+		if skipPos := atomic.LoadInt64(&r.skipPos); skipPos < target {
+			skipInfo := fmt.Sprintf(", skipped %.1f%%", 100*float64(skipPos)/float64(target))
+
+			lastSkipPos := atomic.LoadInt64(&r.lastSkipPos)
+			lastSkipTime := atomic.LoadInt64(&r.lastSkipTime)
+			now := time.Now().Unix()
+
+			if lastSkipTime != 0 && now > lastSkipTime && skipPos > lastSkipPos {
+				lps := float64(skipPos-lastSkipPos) / float64(now-lastSkipTime)
+				remain := time.Duration(float64(target-skipPos) / lps * float64(time.Second))
+				skipInfo += fmt.Sprintf(" remaining %s", remain.Round(time.Second).String())
+			}
+
+			atomic.StoreInt64(&r.lastSkipPos, skipPos)
+			atomic.StoreInt64(&r.lastSkipTime, now)
+
+			res += skipInfo
+		}
+	}
+
 	if r.filters.isSet() || r.options.PrepareLine != nil {
 		m := atomic.LoadInt64(&r.matches)
 		pr.Matches = &m
@@ -187,6 +224,10 @@ func (r *runner) st(s progress.Status) string {
 		}
 	}
 
+	r.statusMu.Lock()
+	r.lastStatus = res
+	r.statusMu.Unlock()
+
 	return res
 }
 
@@ -199,24 +240,52 @@ func (r *runner) readFile(rd io.Reader, out io.Writer) {
 	}
 }
 
+// linesPush picks how many lines to batch before updating shared progress counters,
+// keeping a batch's fill time under the progress status interval so speed/remaining
+// estimates don't alias against a rate-limited throughput.
+func (r *runner) linesPush() int {
+	if r.limiter == nil {
+		return 1000
+	}
+
+	lim := r.limiter.Limit()
+	if lim == rate.Inf || lim > 1000 {
+		return 1000
+	}
+
+	if lim < 1 {
+		return 1
+	}
+
+	return int(lim)
+}
+
 func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer) {
 	s := bufio.NewScanner(rd)
 	s.Buffer(make([]byte, 64*1024), 10*1024*1024)
 
+	atomic.StoreInt64(&r.skipPos, 0)
+	atomic.StoreInt64(&r.lastSkipPos, 0)
+	atomic.StoreInt64(&r.lastSkipTime, 0)
+
 	fileLines := 0
 	lines := 0
 	buf := make([]byte, 64*1024)
 
-	linesPush := 1000
-	if r.rateLimit < 100 {
-		linesPush = 1
-	}
+	linesPush := r.linesPush()
 
 	for s.Scan() {
 		fileLines++
-		lines++
+
+		if atomic.LoadInt64(&r.closed) > 0 {
+			break
+		}
 
 		if r.startLine > 0 && fileLines <= r.startLine {
+			if fileLines&0xFFFF == 0 || fileLines == r.startLine {
+				atomic.StoreInt64(&r.skipPos, int64(fileLines))
+			}
+
 			continue
 		}
 
@@ -224,12 +293,10 @@ func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer) {
 			break
 		}
 
-		if atomic.LoadInt64(&r.closed) > 0 {
-			break
-		}
+		lines++
 
 		if r.limiter != nil {
-			_ = r.limiter.Wait(context.Background()) //nolint:errcheck // No failure condition here.
+			_ = r.limiter.Wait(r.ctx) //nolint:errcheck // No failure condition here.
 		}
 
 		line := s.Bytes()
@@ -243,6 +310,7 @@ func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer) {
 		if lines >= linesPush {
 			atomic.AddInt64(&r.currentLines, int64(lines))
 			lines = 0
+			linesPush = r.linesPush()
 
 			if flusher, ok := w.(interface {
 				Flush() error
@@ -309,6 +377,87 @@ func (r *runner) scanFile(filename string, rd io.Reader, out io.Writer) {
 	}
 }
 
+func formatLimit(lim rate.Limit) string {
+	if lim == rate.Inf {
+		return "unlimited"
+	}
+
+	return strconv.FormatFloat(float64(lim), 'f', -1, 64)
+}
+
+// startControlServer exposes GET/POST /rate-limit to inspect and change the rate limit at runtime.
+func (r *runner) startControlServer() error {
+	ln, err := net.Listen("tcp", r.httpAddr)
+	if err != nil {
+		return fmt.Errorf("failed to listen on %s: %w", r.httpAddr, err)
+	}
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/rate-limit", func(w http.ResponseWriter, req *http.Request) {
+		if req.Method == http.MethodPost {
+			v, err := strconv.ParseFloat(req.FormValue("value"), 64)
+			if err != nil {
+				http.Error(w, err.Error(), http.StatusBadRequest)
+
+				return
+			}
+
+			if v <= 0 {
+				r.limiter.SetLimit(rate.Inf)
+			} else {
+				r.limiter.SetLimit(rate.Limit(v))
+			}
+		}
+
+		fmt.Fprintf(w, "%s\n", formatLimit(r.limiter.Limit()))
+	})
+	mux.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) {
+		r.statusMu.Lock()
+		status := r.lastStatus
+		r.statusMu.Unlock()
+
+		fmt.Fprintf(w, `<html><body>
+<pre>%s</pre>
+<p>rate limit: %s lines/sec</p>
+<form method="post" action="/rate-limit">
+<input name="value" placeholder="lines/sec, 0 or less = unlimited">
+<button type="submit">set</button>
+</form>
+</body></html>`, status, formatLimit(r.limiter.Limit()))
+	})
+
+	var handler http.Handler = mux
+
+	if r.httpAuth != "" {
+		user, pass, _ := strings.Cut(r.httpAuth, ":")
+		handler = basicAuth(user, pass, mux)
+	}
+
+	go func() {
+		if err := http.Serve(ln, handler); err != nil {
+			log.Println("control server stopped:", err.Error())
+		}
+	}()
+
+	return nil
+}
+
+func basicAuth(user, pass string, next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		u, p, ok := req.BasicAuth()
+		if !ok ||
+			subtle.ConstantTimeCompare([]byte(u), []byte(user)) != 1 ||
+			subtle.ConstantTimeCompare([]byte(p), []byte(pass)) != 1 {
+			w.Header().Set("WWW-Authenticate", `Basic realm="catp"`)
+			http.Error(w, "unauthorized", http.StatusUnauthorized)
+
+			return
+		}
+
+		next.ServeHTTP(w, req)
+	})
+}
+
 func (r *runner) cat(filename string) (err error) {
 	var rd io.Reader
 
@@ -390,15 +539,11 @@ func (r *runner) cat(filename string) (err error) {
 		})
 	}
 
-	if r.rateLimit > 0 {
-		r.limiter = rate.NewLimiter(rate.Limit(r.rateLimit), 100)
-	}
-
 	if r.startLine != 0 || r.endLine != 0 {
 		r.countLines = true
 	}
 
-	if r.filters.isSet() || r.parallel > 1 || r.hasOptions || r.countLines || r.rateLimit > 0 {
+	if r.filters.isSet() || r.parallel > 1 || r.hasOptions || r.countLines || r.limiter != nil {
 		r.scanFile(filename, rd, out)
 	} else {
 		r.readFile(rd, out)
diff --git a/go.mod b/go.mod
index 1a15c42..2041398 100644
--- a/go.mod
+++ b/go.mod
@@ -4,7 +4,7 @@ go 1.24.0
 
 require (
 	github.com/DataDog/zstd v1.5.7
-	github.com/bool64/dev v0.2.43
+	github.com/bool64/dev v0.2.45
 	github.com/cloudflare/ahocorasick v0.0.0-20240916140611-054963ec9396
 	github.com/klauspost/compress v1.18.2
 	github.com/klauspost/pgzip v1.2.6
diff --git a/go.sum b/go.sum
index 5d4fe0b..0c780c7 100644
--- a/go.sum
+++ b/go.sum
@@ -1,7 +1,7 @@
 github.com/DataDog/zstd v1.5.7 h1:ybO8RBeh29qrxIhCA9E8gKY6xfONU9T6G6aP9DTKfLE=
 github.com/DataDog/zstd v1.5.7/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=
-github.com/bool64/dev v0.2.43 h1:yQ7qiZVef6WtCl2vDYU0Y+qSq+0aBrQzY8KXkklk9cQ=
-github.com/bool64/dev v0.2.43/go.mod h1:iJbh1y/HkunEPhgebWRNcs8wfGq7sjvJ6W5iabL8ACg=
+github.com/bool64/dev v0.2.45 h1:3nLKhAS/6Oklk3Mt2lHYSN/Cb4tdAD77KLwzeP+6eYE=
+github.com/bool64/dev v0.2.45/go.mod h1:iJbh1y/HkunEPhgebWRNcs8wfGq7sjvJ6W5iabL8ACg=
 github.com/cloudflare/ahocorasick v0.0.0-20240916140611-054963ec9396 h1:W2HK1IdCnCGuLUeyizSCkwvBjdj0ZL7mxnJYQ3poyzI=
 github.com/cloudflare/ahocorasick v0.0.0-20240916140611-054963ec9396/go.mod h1:tGWUZLZp9ajsxUOnHmFFLnqnlKXsCn6GReG4jAD59H0=
 github.com/klauspost/compress v1.18.2 h1:iiPHWW0YrcFgpBYhsA6D1+fqHssJscY/Tm/y2Uqnapk=

From 40a7fb3c3ce2117ad6f8ad47987a9544091fb061 Mon Sep 17 00:00:00 2001
From: Viacheslav Poturaev <vearutop@gmail.com>
Date: Tue, 4 Aug 2026 11:00:30 +0200
Subject: [PATCH 2/2] Fix lint

---
 .golangci.yml                |  9 +++++++++
 cmd/catp/catp/app.go         | 15 ++-------------
 cmd/catp/catp/catp.go        | 10 +++++-----
 cmd/catp/catp/filter_test.go |  4 ++--
 progress.go                  |  2 +-
 5 files changed, 19 insertions(+), 21 deletions(-)

diff --git a/.golangci.yml b/.golangci.yml
index 0a528da..200dd7b 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -5,6 +5,15 @@ run:
 linters:
   default: all
   disable:
+    - cyclop
+    - err113
+    - errcheck
+    - gocognit
+    - gocyclo
+    - funlen
+    - containedctx
+    - nestif
+    - noctx
     - embeddedstructfieldcheck
     - nilnil
     - noinlineerr
diff --git a/cmd/catp/catp/app.go b/cmd/catp/catp/app.go
index 5258f2c..0b155ce 100644
--- a/cmd/catp/catp/app.go
+++ b/cmd/catp/catp/app.go
@@ -13,6 +13,7 @@ import (
 	"os/signal"
 	"path/filepath"
 	"runtime/pprof"
+	"slices"
 	"sort"
 	"strings"
 	"sync/atomic"
@@ -167,17 +168,7 @@ func Main(options ...func(o *Options)) error { //nolint:funlen,cyclop,gocognit,g
 			}
 
 			for _, f := range glob {
-				alreadyThere := false
-
-				for _, e := range files {
-					if e == f {
-						alreadyThere = true
-
-						break
-					}
-				}
-
-				if !alreadyThere {
+				if !slices.Contains(files, f) {
 					files = append(files, f)
 				}
 			}
@@ -284,8 +275,6 @@ func Main(options ...func(o *Options)) error { //nolint:funlen,cyclop,gocognit,g
 		errs := make(chan error, r.parallel)
 
 		for _, fn := range files {
-			fn := fn
-
 			select {
 			case err := <-errs:
 				return err
diff --git a/cmd/catp/catp/catp.go b/cmd/catp/catp/catp.go
index e594ed1..c23a617 100644
--- a/cmd/catp/catp/catp.go
+++ b/cmd/catp/catp/catp.go
@@ -195,7 +195,7 @@ func (r *runner) st(s progress.Status) string {
 			if lastSkipTime != 0 && now > lastSkipTime && skipPos > lastSkipPos {
 				lps := float64(skipPos-lastSkipPos) / float64(now-lastSkipTime)
 				remain := time.Duration(float64(target-skipPos) / lps * float64(time.Second))
-				skipInfo += fmt.Sprintf(" remaining %s", remain.Round(time.Second).String())
+				skipInfo += " remaining " + remain.Round(time.Second).String()
 			}
 
 			atomic.StoreInt64(&r.lastSkipPos, skipPos)
@@ -395,7 +395,7 @@ func (r *runner) startControlServer() error {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/rate-limit", func(w http.ResponseWriter, req *http.Request) {
 		if req.Method == http.MethodPost {
-			v, err := strconv.ParseFloat(req.FormValue("value"), 64)
+			v, err := strconv.ParseFloat(req.FormValue("value"), 64) //nolint:gosec
 			if err != nil {
 				http.Error(w, err.Error(), http.StatusBadRequest)
 
@@ -409,14 +409,14 @@ func (r *runner) startControlServer() error {
 			}
 		}
 
-		fmt.Fprintf(w, "%s\n", formatLimit(r.limiter.Limit()))
+		_, _ = fmt.Fprintf(w, "%s\n", formatLimit(r.limiter.Limit()))
 	})
 	mux.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) {
 		r.statusMu.Lock()
 		status := r.lastStatus
 		r.statusMu.Unlock()
 
-		fmt.Fprintf(w, `<html><body>
+		_, _ = fmt.Fprintf(w, `<html><body>
 <pre>%s</pre>
 <p>rate limit: %s lines/sec</p>
 <form method="post" action="/rate-limit">
@@ -434,7 +434,7 @@ func (r *runner) startControlServer() error {
 	}
 
 	go func() {
-		if err := http.Serve(ln, handler); err != nil {
+		if err := http.Serve(ln, handler); err != nil { //nolint:gosec
 			log.Println("control server stopped:", err.Error())
 		}
 	}()
diff --git a/cmd/catp/catp/filter_test.go b/cmd/catp/catp/filter_test.go
index d34fb69..29c65ce 100644
--- a/cmd/catp/catp/filter_test.go
+++ b/cmd/catp/catp/filter_test.go
@@ -1,4 +1,4 @@
-package catp
+package catp //nolint:testpackage
 
 import (
 	"bytes"
@@ -18,7 +18,7 @@ func TestFilter_Match(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	for _, line := range bytes.Split(input, []byte("\n")) {
+	for line := range bytes.SplitSeq(input, []byte("\n")) {
 		if _, ok := f.shouldWrite(line); ok {
 			println(string(line))
 		}
diff --git a/progress.go b/progress.go
index a3e5902..22ca2a6 100644
--- a/progress.go
+++ b/progress.go
@@ -335,7 +335,7 @@ func (cr *sharedCounters) count(n int, p []byte, err error) {
 		return
 	}
 
-	for i := 0; i < n; i++ {
+	for i := range n {
 		if p[i] == '\n' {
 			cr.localLines++
 		}