opqueue

package
v1.230.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 14, 2026 License: MPL-2.0 Imports: 20 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrQueueFull    = errors.New("operation queue is full")
	ErrQueueStopped = errors.New("operation queue is stopped")
)

Errors

Functions

func GetAllSets 

func GetAllSets() []string

GetAllSets returns all known set names v1.33.0: Dual sets — interval (feeds) + hash (manual)

func GetSourceConfig 

func GetSourceConfig(source string) (sourceConfig, bool)

GetSourceConfig returns the configuration for a source

func GetTargetSet 

func GetTargetSet(source, ip string) string

GetTargetSet returns the appropriate set for a source and IP v1.39.0: CIDRs always route to interval set (hash sets don't support ranges)

func IsAllowedPath 

func IsAllowedPath(path string) bool

IsAllowedPath checks if a path is in allowed directories

func IsPersistedSet added in v1.209.0 

func IsPersistedSet(setName string) bool

IsPersistedSet reports whether a set's element provenance is tracked/persisted (v1.209). The OpQueue apply boundary uses this to decide whether to record source_index provenance.

func SecureStreamElementsFile 

func SecureStreamElementsFile(path string, config StreamingBatchConfig, callback BatchCallback) (elementCount int, err error)

SecureStreamElementsFile opens a file, streams elements in batches via callback, and closes This is a convenience function that handles the full lifecycle Memory usage is O(batch_size) instead of O(n)

func WithTimeout 

func WithTimeout(ctx context.Context, timeout time.Duration) context.Context

WithTimeout returns a context with timeout for operations

Types

type BatchCallback 

type BatchCallback func(batch []string, batchNum int) error

BatchCallback is called for each batch of elements Return error to abort streaming

type BulkJob 

type BulkJob struct {
	SetName  string
	Elements []string
	Source   string
}

BulkJob represents a low-priority bulk operation (replace_set for feeds/geoban)

type FastOp 

type FastOp struct {
	SetName    string
	Op         *SetOp
	EnqueuedAt time.Time // v1.40.0: For enforcement latency tracking
}

FastOp represents a high-priority operation (ban, unban, flush_source, flush_set)

type FileReadResult 

type FileReadResult struct {
	Elements []string
	Count    int
}

FileReadResult contains the result of reading a file

func SecureReadElementsFile 

func SecureReadElementsFile(path string) (*FileReadResult, error)

SecureReadElementsFile reads elements from a file with security checks Security measures: 1. Path prefix validation 2. Symlink rejection (O_NOFOLLOW + Lstat) 3. Regular file check 4. Size limit 5. Ownership check (root or nftban) 6. World-writable check 7. TOCTOU protection (inode verification) 8. Line count limit

type FlushResult 

type FlushResult struct {
	SetName     string
	Applied     int      // TRUE count of elements actually applied (== actually_applied)
	Intended    int      // L2b: elements the replace/flush intended to apply (diagnostic; 0 if N/A)
	Adds        int      // Count of add operations applied (v1.32.0)
	Deletes     int      // Count of delete operations applied (v1.32.0)
	WasReplace  bool     // True if this was a replace_set operation (v1.32.0)
	WasFlush    bool     // True if this was a flush_set operation (v1.32.0)
	PostBarrier []*SetOp // Ops to re-enqueue after flush
	// Err is set when the operation did not fully succeed. L2b invariant:
	// a partial replace (Applied < Intended) is NOT success — Err is non-nil.
	Err error
}

FlushResult contains the outcome of a buffer flush

type IndexEntry 

type IndexEntry struct {
	Set       string   `json:"set"`
	Element   string   `json:"element"`
	Sources   []string `json:"sources"`
	TTL       uint32   `json:"ttl,omitempty"`
	Updated   int64    `json:"updated"`
	ExpiresAt int64    `json:"expires_at,omitempty"` // Unix timestamp; 0 = no expiry
}

IndexEntry represents a persisted index entry

type NFTBackendWrapper 

type NFTBackendWrapper struct {
	// contains filtered or unexported fields
}

NFTBackendWrapper wraps nftsync.NFTManager to implement NetlinkBackend

func NewNFTBackendWrapper 

func NewNFTBackendWrapper(nft *nftsync.NFTManager) (*NFTBackendWrapper, error)

NewNFTBackendWrapper creates a wrapper around an existing NFTManager

func (*NFTBackendWrapper) AddElements 

func (w *NFTBackendWrapper) AddElements(tableName, setName string, elements []SetElement) (int, error)

AddElements adds elements to a set (batched). L2b (OPQUEUE_PER_ELEMENT_APPLY_RESULT): truth-bearing — returns the count ACTUALLY applied and, if any element failed, a non-nil error carrying the applied/total shortfall + the first failure. Previously this swallowed per-element failures and returned nil (success-shaped), which hid partial replace_set applies from the caller. It still continues past a single bad element (so one bad IP does not abort the whole batch), but no longer reports that as full success.

func (*NFTBackendWrapper) Close 

func (w *NFTBackendWrapper) Close()

Close is a no-op since we don't own the NFTManager

func (*NFTBackendWrapper) DeleteElements 

func (w *NFTBackendWrapper) DeleteElements(tableName, setName string, elements []SetElement) error

DeleteElements removes elements from a set (batched)

func (*NFTBackendWrapper) FlushSet 

func (w *NFTBackendWrapper) FlushSet(tableName, setName string) error

FlushSet clears all elements from a set

func (*NFTBackendWrapper) GetSetCount 

func (w *NFTBackendWrapper) GetSetCount(tableName, setName string) (int, error)

GetSetCount returns the number of elements in a set without loading them all into memory. For interval sets, filters out IntervalEnd markers to return the true entry count.

func (*NFTBackendWrapper) GetSetElements 

func (w *NFTBackendWrapper) GetSetElements(tableName, setName string) ([]string, error)

GetSetElements returns all elements in a set

func (*NFTBackendWrapper) InvalidateCache 

func (w *NFTBackendWrapper) InvalidateCache()

InvalidateCache clears the set cache (call after external nft changes)

func (*NFTBackendWrapper) ReplaceSet added in v1.221.0 

func (w *NFTBackendWrapper) ReplaceSet(tableName, setName string, elements []SetElement) error

ReplaceSet atomically replaces the entire contents of a set in ONE netlink transaction, delegating to the shared NFTManager connection. It does NOT call the separately-committing FlushSet()/AddElements() methods — flush + add + commit happen as a single transaction, so the set is never transiently empty (fail-CLOSED on failure). The address family is derived from the resolved set's KeyType inside ReplaceSetElements, never from setName.

type NetlinkBackend 

type NetlinkBackend interface {
	// FlushSet clears all elements from a set
	FlushSet(table, set string) error

	// AddElements adds elements to a set (batched). L2b: truth-bearing contract —
	// returns the count ACTUALLY applied and a non-nil error if any element failed
	// (applied < len(elements)). The returned count MUST equal what was really applied;
	// callers rely on partial_apply != success.
	AddElements(table, set string, elements []SetElement) (applied int, err error)

	// ReplaceSet atomically replaces the ENTIRE contents of a set in ONE netlink
	// transaction (flush + add + single commit). It is the atomic primitive for the
	// replace_set / flush_source paths: on any failure NOTHING commits and the set
	// retains its prior contents (fail-CLOSED), never the transient-empty fail-OPEN
	// of a standalone FlushSet()-then-AddElements(). Elements are validated before
	// the connection is mutated; an empty slice is an explicit atomic flush. Callers
	// MUST route replacement through this and MUST NOT call FlushSet()+AddElements()
	// separately for a replacement.
	ReplaceSet(table, set string, elements []SetElement) error

	// DeleteElements removes elements from a set (batched)
	DeleteElements(table, set string, elements []SetElement) error

	// GetSetElements returns all elements in a set
	GetSetElements(table, set string) ([]string, error)
}

NetlinkBackend interface for nftables operations (allows mocking in tests)

type OpQueue 

type OpQueue struct {
	// contains filtered or unexported fields
}

OpQueue manages all set buffers with async flush workers

func NewOpQueue 

func NewOpQueue(backend NetlinkBackend, config QueueConfig) *OpQueue

NewOpQueue creates a new operation queue

func (*OpQueue) CheckExempt added in v1.218.0 

func (q *OpQueue) CheckExempt(set, ip string) (bool, string)

CheckExempt reports whether adding ip into set would violate the never-ban invariant, using the injected resolver. Nil-safe (no resolver → not exempt). Exposed so ban-path callers (e.g. the BotGuard enforcer) can pre-check and skip with a clear log.

func (*OpQueue) EnqueueBan 

func (q *OpQueue) EnqueueBan(setName string, element string, ttl uint32, source, reason string) error

EnqueueBan adds a ban operation (async, non-blocking)

func (*OpQueue) EnqueueFlush 

func (q *OpQueue) EnqueueFlush(setName string) error

EnqueueFlush adds a flush_set operation (async, non-blocking)

func (*OpQueue) EnqueueReplace 

func (q *OpQueue) EnqueueReplace(setName string, elements []string, source string) error

EnqueueReplace adds a replace_set operation (async, non-blocking)

func (*OpQueue) EnqueueReplaceFromFile 

func (q *OpQueue) EnqueueReplaceFromFile(setName, filePath, source string, batchConfig StreamingBatchConfig) (int, error)

EnqueueReplaceFromFile performs a streaming replace_set operation Instead of loading all elements into memory, this: 1. Flushes the set immediately 2. Reads the file in batches and adds elements directly Memory usage is O(batch_size) instead of O(n)

This is a SYNCHRONOUS operation (unlike EnqueueReplace) because it needs to stream from the file. The file can be deleted after return.

func (*OpQueue) EnqueueUnban 

func (q *OpQueue) EnqueueUnban(setName string, element string, source string) error

EnqueueUnban adds an unban operation (async, non-blocking)

func (*OpQueue) QueueDepth 

func (q *OpQueue) QueueDepth() int64

QueueDepth returns total pending operations - O(1), no locks

func (*OpQueue) SetExemptResolver added in v1.218.0 

func (q *OpQueue) SetExemptResolver(fn func(set, ip string) (bool, string))

SetExemptResolver injects the L3b never-ban guard. fn reports (exempt, reason) for adding ip into set; it should combine the enforcement-set classifier with the authoritative IsExempt resolver (same one backend.Ban / L3a use). Safe to call once at daemon init. nil disables the guard (fail-safe).

func (*OpQueue) SetOnFlush 

func (q *OpQueue) SetOnFlush(fn func(setName string, applied int, opType string))

SetOnFlush registers a callback that fires after each successful flush. Callback receives: setName, count of elements applied, operation type. Used by set counters to maintain in-memory element counts (v1.32.0).

func (*OpQueue) SetSourceIndexer added in v1.209.0 

func (q *OpQueue) SetSourceIndexer(si SourceIndexer)

SetSourceIndexer wires the provenance recorder. MUST be called during daemon init BEFORE Start()/any flush so every apply records provenance.

func (*OpQueue) Start 

func (q *OpQueue) Start(ctx context.Context)

Start begins the async flush worker

func (*OpQueue) Stats 

func (q *OpQueue) Stats() QueueStats

Stats returns queue statistics

func (*OpQueue) Stop 

func (q *OpQueue) Stop()

Stop signals the queue to stop and drains remaining operations

type OpType 

type OpType int

OpType represents the type of operation 

const (
	OpAdd OpType = iota
	OpDelete
	OpReplaceSet
	OpFlushSet
)

func (OpType) String 

func (t OpType) String() string

type PendingOp 

type PendingOp struct {
	Type      OpType
	Element   string
	TTL       uint32 // Seconds, 0 = permanent (infinity in comparisons)
	Source    string // Latest source (for logging)
	Reason    string // Latest reason (for logging)
	CreatedAt time.Time
}

PendingOp represents a coalesced pending operation

type QueueConfig 

type QueueConfig struct {
	// Flush policies
	FlushInterval  time.Duration // Time-based flush interval (default: 100ms)
	FlushThreshold int           // Size-based flush threshold (default: 1000 ops)
	MaxBatchSize   int           // Max elements per netlink batch (default: 5000)
	MaxQueueDepth  int64         // Backpressure limit (default: 50000)
}

QueueConfig holds configuration for the operation queue

func DefaultQueueConfig 

func DefaultQueueConfig() QueueConfig

DefaultQueueConfig returns sensible defaults

type QueueStats 

type QueueStats struct {
	PendingCount int64
	TotalQueued  uint64
	TotalApplied uint64
	TotalDropped uint64
	// L2b: number of replace_set applies that flushed then applied fewer elements than
	// intended (partial/fail-open). Non-zero = degraded; the set is short of requested.
	ReplacePartialFailures uint64
	// L3b: number of EnqueueBan calls refused because a single exempt IP targeted an
	// enforcement/drop set (never-ban invariant on the opqueue ban path).
	EnqueueBanExemptSkips uint64
	LastFlushTime         time.Time
}

QueueStats holds queue statistics

type Scheduler 

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler manages priority-based operation scheduling with two lanes

func NewScheduler 

func NewScheduler(backend NetlinkBackend, config QueueConfig) *Scheduler

NewScheduler creates a new priority scheduler v1.19.28: Increased bulkCh from 100 to 1000 to prevent saturation during feed sync

func (*Scheduler) EnqueueBulk 

func (s *Scheduler) EnqueueBulk(setName string, elements []string, source string) error

EnqueueBulk adds a low-priority bulk operation (replace_set)

func (*Scheduler) EnqueueBulkFromFile 

func (s *Scheduler) EnqueueBulkFromFile(ctx context.Context, setName, filePath, source string, batchConfig StreamingBatchConfig) (int, error)

EnqueueBulkFromFile performs a streaming replace_set operation Instead of loading all elements into memory, this: 1. Flushes the set immediately 2. Reads the file in batches and adds elements directly Memory usage is O(batch_size) instead of O(n)

This is a SYNCHRONOUS operation because it needs to stream from the file. The file can be deleted after this function returns.

func (*Scheduler) EnqueueFast 

func (s *Scheduler) EnqueueFast(setName string, op *SetOp) error

EnqueueFast adds a high-priority operation (ban, unban, flush_source, flush_set)

func (*Scheduler) SchedulerStats 

func (s *Scheduler) SchedulerStats() SchedulerStats

SchedulerStats returns scheduler statistics

func (*Scheduler) Start 

func (s *Scheduler) Start(ctx context.Context)

Start begins the scheduler workers

func (*Scheduler) Stop 

func (s *Scheduler) Stop()

Stop signals the scheduler to stop and waits for workers to finish

type SchedulerStats 

type SchedulerStats struct {
	FastPending  int64
	BulkPending  int64
	TotalApplied uint64
	TotalDropped uint64
	// v1.19.28: Per-lane drop counters for granular observability
	FastDropped   uint64
	BulkDropped   uint64
	LastFastFlush time.Time
	LastBulkFlush time.Time
}

SchedulerStats holds scheduler statistics

type SetBuffer 

type SetBuffer struct {
	// contains filtered or unexported fields
}

SetBuffer holds pending operations for one set with coalescing

type SetElement 

type SetElement struct {
	Value  string
	TTL    uint32
	IsIPv6 bool
}

SetElement represents an element to add/delete from nftables

type SetOp 

type SetOp struct {
	Type     OpType
	Element  string   // IP or CIDR (for Add/Delete)
	Elements []string // For ReplaceSet
	TTL      uint32   // Seconds, 0 = permanent
	Reason   string   // For logging
	Source   string   // Origin module (login, portscan, ddos, etc.)
}

SetOp represents a single operation to be queued

type SourceIndex 

type SourceIndex struct {
	// contains filtered or unexported fields
}

SourceIndex tracks which elements came from which source Enables flush_source without dedicated sets per module

func NewSourceIndex 

func NewSourceIndex(indexPath string) *SourceIndex

NewSourceIndex creates a new source index

func (*SourceIndex) Add 

func (si *SourceIndex) Add(setName, element, source string)

Add adds an element with its source

func (*SourceIndex) AddWithExpiry added in v1.38.0 

func (si *SourceIndex) AddWithExpiry(setName, element, source string, expiresAt int64)

AddWithExpiry adds an element with its source and an expiry time. If expiresAt is 0, the entry never expires.

func (*SourceIndex) ClearSet 

func (si *SourceIndex) ClearSet(setName string)

ClearSet removes all entries for a set

func (*SourceIndex) GetAllElements 

func (si *SourceIndex) GetAllElements(setName string) map[string]bool

GetAllElements returns all elements in a set (as a map for O(1) lookup)

func (*SourceIndex) GetElementsBySource 

func (si *SourceIndex) GetElementsBySource(setName, source string) []string

GetElementsBySource returns all elements in a set from a specific source

func (*SourceIndex) HasElement 

func (si *SourceIndex) HasElement(setName, element string) bool

HasElement checks if an element exists in a set

func (*SourceIndex) LoadFromDisk 

func (si *SourceIndex) LoadFromDisk() error

LoadFromDisk restores index on daemon startup

func (*SourceIndex) PruneExpired added in v1.38.0 

func (si *SourceIndex) PruneExpired() int

PruneExpired removes entries whose ExpiresAt has passed. Returns the number of entries pruned.

func (*SourceIndex) ReconcileWithBackend 

func (si *SourceIndex) ReconcileWithBackend(backend NetlinkBackend)

ReconcileWithBackend syncs index with actual nft state Uses O(n) map lookups instead of O(n²) nested loops v1.33.0: Both interval and hash blacklist sets need reconciliation

func (*SourceIndex) Remove 

func (si *SourceIndex) Remove(setName, element, source string)

Remove removes a source from an element

func (*SourceIndex) RemoveAll 

func (si *SourceIndex) RemoveAll(setName, element string)

RemoveAll removes all sources from an element

func (*SourceIndex) SaveToDisk 

func (si *SourceIndex) SaveToDisk() error

SaveToDisk persists index (called after netlink apply, not on enqueue)

func (*SourceIndex) ShouldDelete 

func (si *SourceIndex) ShouldDelete(setName, element string) bool

ShouldDelete returns true if element has no remaining sources

func (*SourceIndex) StartBackgroundSaver 

func (si *SourceIndex) StartBackgroundSaver(ctx context.Context)

StartBackgroundSaver runs periodic persistence

func (*SourceIndex) Stop 

func (si *SourceIndex) Stop()

Stop signals the background saver to stop

type SourceIndexer added in v1.209.0 

type SourceIndexer interface {
	AddWithExpiry(setName, element, source string, expiresAt int64)
}

SourceIndexer records element provenance at the durable apply boundary (v1.209). The OpQueue depends on this interface — not on any producing module — so wiring is lifecycle-race-free. *SourceIndex satisfies it.

type StandaloneBackend 

type StandaloneBackend struct {
	*NFTBackendWrapper
	// contains filtered or unexported fields
}

StandaloneBackend creates its own NFTManager connection Use this only for testing or when no existing Backend is available

func NewStandaloneBackend 

func NewStandaloneBackend() (*StandaloneBackend, error)

NewStandaloneBackend creates a backend with its own netlink connection

func (*StandaloneBackend) Close 

func (s *StandaloneBackend) Close()

Close closes the owned NFTManager connection

type StreamingBatchConfig 

type StreamingBatchConfig struct {
	BatchSize int // Elements per batch (default: 10000)
}

StreamingBatchConfig configures the streaming file reader

func DefaultStreamingBatchConfig 

func DefaultStreamingBatchConfig() StreamingBatchConfig

DefaultStreamingBatchConfig returns sensible defaults

type StreamingFileReader 

type StreamingFileReader struct {
	// contains filtered or unexported fields
}

StreamingFileReader provides streaming access to element files It maintains an open file handle and reads elements in batches

func SecureOpenElementsFile 

func SecureOpenElementsFile(path string, config StreamingBatchConfig) (*StreamingFileReader, error)

SecureOpenElementsFile opens a file with security checks for streaming Returns a reader that must be closed after use Security measures are identical to SecureReadElementsFile

func (*StreamingFileReader) Close 

func (r *StreamingFileReader) Close() error

Close closes the underlying file

func (*StreamingFileReader) ReadNextBatch 

func (r *StreamingFileReader) ReadNextBatch() ([]string, error)

ReadNextBatch reads the next batch of elements Returns the batch (may be smaller than BatchSize at end of file) Returns nil, nil when EOF is reached Returns nil, error on read error or limit exceeded

func (*StreamingFileReader) Stats 

func (r *StreamingFileReader) Stats() (lineCount, elementCount int)

Stats returns reading statistics

func (*StreamingFileReader) StreamElements 

func (r *StreamingFileReader) StreamElements(callback BatchCallback) error

StreamElements reads elements in batches and calls the callback for each This is the main streaming API - memory stays O(batch_size)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.