From 3292f77e5cb32ac56d8747eff986ce4ecf561e99 Mon Sep 17 00:00:00 2001 From: PTah Date: Mon, 13 Apr 2026 15:41:16 +1000 Subject: [PATCH] =?UTF-8?q?fix:=20=D0=BE=D1=82=D0=BC=D0=B5=D0=BD=D0=B0=20?= =?UTF-8?q?=D1=81=D0=BA=D0=B0=D0=BD=D0=B0=20=D0=BF=D1=80=D0=B8=20=D0=B8?= =?UTF-8?q?=D1=81=D1=87=D0=B5=D1=80=D0=BF=D0=B0=D0=BD=D0=B8=D0=B8=20=D1=81?= =?UTF-8?q?=D0=BE=D0=B5=D0=B4=D0=B8=D0=BD=D0=B5=D0=BD=D0=B8=D0=B9=20Postgr?= =?UTF-8?q?es?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Распознавание 53300/too many clients и др. в isStoreCapacityOrConnExhausted - SaveHostResult/порты/SNMP: при фатальной ошибке БД — CancelScan; ctx отменяет дальнейший SNMP - Воркеры при ctx.Done быстро скипают оставшиеся IP без SNMP (дренаж канала) Made-with: Cursor --- internal/scans/runner.go | 74 ++++++++++++++++++++------- internal/scans/store_db_fatal.go | 20 ++++++++ internal/scans/store_db_fatal_test.go | 21 ++++++++ 3 files changed, 97 insertions(+), 18 deletions(-) create mode 100644 internal/scans/store_db_fatal.go create mode 100644 internal/scans/store_db_fatal_test.go diff --git a/internal/scans/runner.go b/internal/scans/runner.go index be4d424..e24d91b 100644 --- a/internal/scans/runner.go +++ b/internal/scans/runner.go @@ -174,6 +174,16 @@ func (r *Runner) run(ctx context.Context, job ScanJob) { const minProgressWriteInterval = 500 * time.Millisecond var progWriteMu sync.Mutex var lastProgressWrite time.Time + var storeFatalOnce sync.Once + storeAbortIfFatal := func(op string, err error) { + if err == nil || !isStoreCapacityOrConnExhausted(err) { + return + } + storeFatalOnce.Do(func() { + log.Printf("scan %s: критическая ошибка БД (%s): %v — отмена скана", job.ID, op, err) + r.CancelScan(job.ID) + }) + } pushProgress := func(force bool) { mu.Lock() @@ -207,21 +217,34 @@ func (r *Runner) run(ctx context.Context, job ScanJob) { worker := func() { defer wg.Done() for ip := range workCh { + if ctx.Err() != nil { + continue + } isUp := probeHost(ip, job.Options.PingTimeoutMS) checkedAt := time.Now().UTC() - _ = r.store.SaveHostResult(job.ID, HostResult{ + if err := r.store.SaveHostResult(job.ID, HostResult{ IP: ip, IsUp: isUp, CheckedAt: checkedAt, - }) - if isUp && job.Options.PortScanEnabled { + }); err != nil { + storeAbortIfFatal("SaveHostResult", err) + if !isStoreCapacityOrConnExhausted(err) { + log.Printf("scan %s: SaveHostResult %s: %v", job.ID, ip, err) + } + } else if isUp && job.Options.PortScanEnabled { for _, p := range job.Options.Ports { - _ = r.store.SaveOpenPortResult(job.ID, OpenPortResult{ + if err := r.store.SaveOpenPortResult(job.ID, OpenPortResult{ IP: ip, Port: p, IsOpen: probeTCPPort(ip, p, job.Options.PingTimeoutMS), CheckedAt: checkedAt, - }) + }); err != nil { + storeAbortIfFatal("SaveOpenPortResult", err) + if !isStoreCapacityOrConnExhausted(err) { + log.Printf("scan %s: SaveOpenPortResult %s:%d: %v", job.ID, ip, p, err) + } + break + } } } @@ -233,22 +256,37 @@ func (r *Runner) run(ctx context.Context, job ScanJob) { mu.Unlock() pushProgress(false) - if isUp && r.cfg.SNMPEnabled { + if isUp && r.cfg.SNMPEnabled && ctx.Err() == nil { snmpRes, lldpRes, ifRes, portDevRes := probeSNMPAndLLDP(ip, r.cfg.SNMPCommunity, checkedAt, job.Options.PingTimeoutMS) - _ = r.store.SaveSNMPResult(job.ID, snmpRes) - for _, lr := range lldpRes { - if err := r.store.SaveLLDPResult(job.ID, lr); err != nil { - log.Printf("scan %s: save LLDP для %s: %v", job.ID, lr.IP, err) + if err := r.store.SaveSNMPResult(job.ID, snmpRes); err != nil { + storeAbortIfFatal("SaveSNMPResult", err) + if !isStoreCapacityOrConnExhausted(err) { + log.Printf("scan %s: SaveSNMPResult %s: %v", job.ID, ip, err) } - } - for _, iface := range ifRes { - if err := r.store.SaveInterfaceResult(job.ID, iface); err != nil { - log.Printf("scan %s: save interface %s ifIndex=%d: %v", job.ID, iface.IP, iface.IfIndex, err) + } else { + for _, lr := range lldpRes { + if err := r.store.SaveLLDPResult(job.ID, lr); err != nil { + storeAbortIfFatal("SaveLLDPResult", err) + if !isStoreCapacityOrConnExhausted(err) { + log.Printf("scan %s: save LLDP для %s: %v", job.ID, lr.IP, err) + } + } } - } - for _, pd := range portDevRes { - if err := r.store.SavePortDeviceResult(job.ID, pd); err != nil { - log.Printf("scan %s: save port-device %s ifIndex=%d mac=%s: %v", job.ID, pd.IP, pd.IfIndex, pd.MAC, err) + for _, iface := range ifRes { + if err := r.store.SaveInterfaceResult(job.ID, iface); err != nil { + storeAbortIfFatal("SaveInterfaceResult", err) + if !isStoreCapacityOrConnExhausted(err) { + log.Printf("scan %s: save interface %s ifIndex=%d: %v", job.ID, iface.IP, iface.IfIndex, err) + } + } + } + for _, pd := range portDevRes { + if err := r.store.SavePortDeviceResult(job.ID, pd); err != nil { + storeAbortIfFatal("SavePortDeviceResult", err) + if !isStoreCapacityOrConnExhausted(err) { + log.Printf("scan %s: save port-device %s ifIndex=%d mac=%s: %v", job.ID, pd.IP, pd.IfIndex, pd.MAC, err) + } + } } } } diff --git a/internal/scans/store_db_fatal.go b/internal/scans/store_db_fatal.go new file mode 100644 index 0000000..e1bf5d1 --- /dev/null +++ b/internal/scans/store_db_fatal.go @@ -0,0 +1,20 @@ +package scans + +import ( + "strings" +) + +// isStoreCapacityOrConnExhausted — ошибки, при которых дальнейший проход скана в БД бессмысленен (слоты Postgres, пул и т.п.). +func isStoreCapacityOrConnExhausted(err error) bool { + if err == nil { + return false + } + msg := strings.ToLower(err.Error()) + return strings.Contains(msg, "53300") || + strings.Contains(msg, "too many clients") || + strings.Contains(msg, "remaining connection slots") || + strings.Contains(msg, "connection slots are reserved") || + strings.Contains(msg, "sql: connection is closed") || + strings.Contains(msg, "broken pipe") || + strings.Contains(msg, "connection reset by peer") +} diff --git a/internal/scans/store_db_fatal_test.go b/internal/scans/store_db_fatal_test.go new file mode 100644 index 0000000..43de8a6 --- /dev/null +++ b/internal/scans/store_db_fatal_test.go @@ -0,0 +1,21 @@ +package scans + +import ( + "errors" + "testing" +) + +func TestIsStoreCapacityOrConnExhausted(t *testing.T) { + if !isStoreCapacityOrConnExhausted(errors.New(`FATAL: sorry, too many clients already (SQLSTATE 53300)`)) { + t.Fatal("expected too many clients") + } + if !isStoreCapacityOrConnExhausted(errors.New(`remaining connection slots are reserved for roles`)) { + t.Fatal("expected reserved slots") + } + if isStoreCapacityOrConnExhausted(errors.New("duplicate key value violates unique constraint")) { + t.Fatal("unexpected match for unrelated error") + } + if isStoreCapacityOrConnExhausted(nil) { + t.Fatal("nil should be false") + } +}