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") + } +}