refactor: enhance bird metrics polling with context support

Updated the `startBirdMetricsPoller` function to accept a context parameter, allowing for better control over the polling lifecycle. This change was applied in both `evobgp-all` and `evobgp-api` main files. Additionally, modified the `StartBirdProtocolsPoller` function to handle context cancellation, ensuring graceful shutdown of the polling routine. Introduced a new service in the Docker Compose configuration for logging runtime service outputs, improving observability during deployment.

internal/jobs/job.go

@@ -2,7 +2,10 @@ package jobs
 
 import (
 	"fmt"
+	"os"
 	"sort"
+	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -20,17 +23,17 @@ const (
 
 // Job is the API-facing job model (поля согласованы со схемой job_audit в миграциях; персистенция в БД пока не подключена).
 type Job struct {
-	ID               string
-	TenantID         string
-	Kind             string
-	Status           string
-	IdempotencyKey   *string
-	ModuleID         *string
-	CreatedAt        time.Time
-	StartedAt        *time.Time
-	FinishedAt       *time.Time
-	Error            *string
-	ProgressPct      *int16
+	ID              string
+	TenantID        string
+	Kind            string
+	Status          string
+	IdempotencyKey  *string
+	ModuleID        *string
+	CreatedAt       time.Time
+	StartedAt       *time.Time
+	FinishedAt      *time.Time
+	Error           *string
+	ProgressPct     *int16
 	Meta            map[string]any
 	cancelRequested bool
 	mu              sync.Mutex
@@ -148,6 +151,25 @@ func (j *Job) Snapshot() map[string]any {
 	return m
 }
 
+var jobRegistryMaxJobsOnce sync.Once
+var jobRegistryMaxJobs int
+
+// registryMaxJobsFromEnv returns EVOBGP_JOB_REGISTRY_MAX_JOBS once (0 = без лимита, только завершённые джобы вытесняются).
+func registryMaxJobsFromEnv() int {
+	jobRegistryMaxJobsOnce.Do(func() {
+		s := strings.TrimSpace(os.Getenv("EVOBGP_JOB_REGISTRY_MAX_JOBS"))
+		if s == "" {
+			return
+		}
+		n, err := strconv.Atoi(s)
+		if err != nil || n <= 0 {
+			return
+		}
+		jobRegistryMaxJobs = n
+	})
+	return jobRegistryMaxJobs
+}
+
 // Registry — in-memory очередь и индекс по idempotency в процессе, где поднят HTTP API (evobgp-api и evobgp-all).
 // Отдельные воркеры в reference-профиле не разделяют память с API: scheduler дергает refresh по HTTP; см. docs/architecture.md.
 // Запись задач в PostgreSQL job_audit + SKIP LOCKED / внешний брокер — планируемое расширение (архитектурный план §2, §7.10).
@@ -171,11 +193,54 @@ func NewRegistry(workerStart func(j *Job)) *Registry {
 	}
 }
 
+// pruneTerminalIfOver удаляет самые старые завершённые джобы (succeeded/failed/cancelled), пока len(byID) > maxJobs.
+func (r *Registry) pruneTerminalIfOver(maxJobs int) {
+	if r == nil || maxJobs <= 0 || len(r.byID) <= maxJobs {
+		return
+	}
+	type fin struct {
+		j *Job
+		t time.Time
+	}
+	var cands []fin
+	for _, j := range r.byID {
+		st := j.statusLocked()
+		if st != StatusSucceeded && st != StatusFailed && st != StatusCancelled {
+			continue
+		}
+		j.mu.Lock()
+		ft := j.FinishedAt
+		j.mu.Unlock()
+		if ft == nil {
+			continue
+		}
+		cands = append(cands, fin{j: j, t: *ft})
+	}
+	need := len(r.byID) - maxJobs
+	if need <= 0 || len(cands) == 0 {
+		return
+	}
+	sort.Slice(cands, func(i, j int) bool { return cands[i].t.Before(cands[j].t) })
+	if need > len(cands) {
+		need = len(cands)
+	}
+	for i := 0; i < need; i++ {
+		v := cands[i].j
+		delete(r.byID, v.ID)
+		if v.IdempotencyKey != nil && *v.IdempotencyKey != "" {
+			delete(r.byIdempo, idempoKey{tenant: v.TenantID, key: *v.IdempotencyKey})
+		}
+	}
+}
+
 // Enqueue creates a job or returns an existing one for the same idempotency key.
 func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, moduleID *string, meta map[string]any) (*Job, bool, error) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
+	maxJobs := registryMaxJobsFromEnv()
+	r.pruneTerminalIfOver(maxJobs)
+
 	if idempotencyKey != nil && *idempotencyKey != "" {
 		k := idempoKey{tenant: tenantID, key: *idempotencyKey}
 		if existing, ok := r.byIdempo[k]; ok {
@@ -197,6 +262,7 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module
 		r.byIdempo[idempoKey{tenant: tenantID, key: *idempotencyKey}] = j
 	}
 	r.byID[j.ID] = j
+	r.pruneTerminalIfOver(maxJobs)
 
 	if r.workerStart != nil {
 		go r.workerStart(j)

internal/jobs/worker.go

@@ -158,8 +158,8 @@ func (w *Worker) finishModuleRefreshSuccess(j *Job, rev string) {
 	deferDeploy := w.Registry.CountOtherActiveModuleRefresh(j.TenantID, j.ID) > 0
 	if deferDeploy {
 		j.mergeMeta(map[string]any{
-			"deploy_apply_deferred":        true,
-			"deploy_apply_defer_reason":    "parallel_module_refresh",
+			"deploy_apply_deferred":     true,
+			"deploy_apply_defer_reason": "parallel_module_refresh",
 		})
 	}
 	j.Succeed()
@@ -324,19 +324,6 @@ func (w *Worker) buildRevisionLogEntries(tenantID, revID string) ([]map[string]a
 		return nil, 0, fmt.Errorf("store not configured")
 	}
 	commLabels := buildCommunityLabelMap(w.Store, tenantID)
-	var all []store.PrefixRow
-	cursor := ""
-	for {
-		rows, next, more := w.Store.ListRevisionPrefixes(tenantID, revID, cursor, 1000)
-		all = append(all, rows...)
-		if !more {
-			break
-		}
-		cursor = next
-		if strings.TrimSpace(cursor) == "" {
-			break
-		}
-	}
 	type agg struct {
 		kind      string
 		source    string
@@ -345,22 +332,35 @@ func (w *Worker) buildRevisionLogEntries(tenantID, revID string) ([]map[string]a
 		sample    []string
 	}
 	groups := map[string]*agg{}
-	for _, p := range all {
-		src := strings.TrimSpace(p.Source)
-		comm := "none"
-		if p.CommunityID != nil && strings.TrimSpace(*p.CommunityID) != "" {
-			comm = strings.TrimSpace(*p.CommunityID)
+	total := 0
+	cursor := ""
+	for {
+		rows, next, more := w.Store.ListRevisionPrefixes(tenantID, revID, cursor, 1000)
+		for _, p := range rows {
+			total++
+			src := strings.TrimSpace(p.Source)
+			comm := "none"
+			if p.CommunityID != nil && strings.TrimSpace(*p.CommunityID) != "" {
+				comm = strings.TrimSpace(*p.CommunityID)
+			}
+			kind, sourceName := classifySource(src)
+			k := kind + "|" + sourceName + "|" + comm
+			g, ok := groups[k]
+			if !ok {
+				g = &agg{kind: kind, source: sourceName, community: comm}
+				groups[k] = g
+			}
+			g.count++
+			if len(g.sample) < 5 {
+				g.sample = append(g.sample, p.Prefix)
+			}
 		}
-		kind, sourceName := classifySource(src)
-		k := kind + "|" + sourceName + "|" + comm
-		g, ok := groups[k]
-		if !ok {
-			g = &agg{kind: kind, source: sourceName, community: comm}
-			groups[k] = g
+		if !more {
+			break
 		}
-		g.count++
-		if len(g.sample) < 5 {
-			g.sample = append(g.sample, p.Prefix)
+		cursor = next
+		if strings.TrimSpace(cursor) == "" {
+			break
 		}
 	}
 	keys := make([]string, 0, len(groups))
@@ -374,16 +374,16 @@ func (w *Worker) buildRevisionLogEntries(tenantID, revID string) ([]map[string]a
 		cl := resolveCommunityLabel(g.community, commLabels)
 		msg := humanLogMessage(g.kind, g.source, g.count, cl, g.sample)
 		out = append(out, map[string]any{
-			"kind":             g.kind,
-			"source":           g.source,
-			"community":        g.community,
-			"community_label":  cl,
-			"prefix_count":     g.count,
-			"sample":           g.sample,
-			"message":          msg,
+			"kind":            g.kind,
+			"source":          g.source,
+			"community":       g.community,
+			"community_label": cl,
+			"prefix_count":    g.count,
+			"sample":          g.sample,
+			"message":         msg,
 		})
 	}
-	return out, len(all), nil
+	return out, total, nil
 }
 
 func classifySource(src string) (kind, name string) {