poster

type DeferredCheckError ¶ added in v0.0.29

type DeferredCheckError struct {
	FailedArticles []FailedArticleInfo
	TotalArticles  int
}

DeferredCheckError is a non-fatal error indicating some articles need deferred verification. The upload itself succeeded, but article verification (STAT check) failed after all immediate retries.

func (e *DeferredCheckError) Error() string

type Engine ¶ added in v0.0.30

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

Engine is the process-wide upload resource owner. It bounds the number of articles posted concurrently (worker slots) and the total memory reserved for in-flight article buffers (buffer budget), independent of how many queue jobs are active. One Engine is shared by every job through the transfer runtime.

All methods are safe for concurrent use and safe to call on a nil *Engine, in which case they are no-ops (preserving standalone behaviour when no engine is injected).

func NewEngine(articleSize uint64, bufferLimit int64, connCapacity int) *Engine

func (e *Engine) AcquireWorker(ctx context.Context) error

func (e *Engine) Budget() EngineBudget

func (e *Engine) Metrics() Metrics

func (e *Engine) PerArticleBytes() int64

func (e *Engine) ReleaseBuffer(n int64)

func (e *Engine) ReleaseWorker()

func (e *Engine) ReserveBuffer(ctx context.Context, n int64) error

type EngineBudget ¶ added in v0.0.30

type EngineBudget struct {
	// PerArticleBytes is the memory reserved for a single in-flight article
	// (raw body + estimated encoded size + fixed overhead).
	PerArticleBytes int64
	// BudgetBytes is the total buffer memory the engine may reserve at once.
	BudgetBytes int64
	// WorkerCount is the maximum number of articles posted concurrently.
	WorkerCount int64
}

EngineBudget is the resolved set of process-wide upload limits.

func ComputeEngineBudget(articleSize uint64, bufferLimit int64, connCapacity int) EngineBudget

per_article          = article_size + estimated_yenc_size + 256 KiB
auto_budget          = clamp(per_article * min(conn_capacity, 32), 64MiB, 512MiB)
worker_count         = min(conn_capacity, floor(budget / per_article))

type FailedArticleInfo ¶ added in v0.0.29

type FailedArticleInfo struct {
	MessageID string
	Groups    []string
}

FailedArticleInfo contains information about an article that failed verification and should be deferred for later checking.

type ManifestSink ¶ added in v0.0.30

type ManifestSink interface {
	RecordFile(ctx context.Context, filePath string, articles []*article.Article) error
	// ExistingArticles returns a previously recorded manifest's article records
	// for filePath (ok=false if none), used for crash recovery to reuse the
	// exact Message-IDs rather than regenerating them.
	ExistingArticles(ctx context.Context, filePath string) ([]manifest.ArticleRecord, bool, error)
}

ManifestSink records a durable manifest for a file's articles before they are posted. It is implemented outside the poster (by the transfer runtime) and injected per job; a nil sink disables manifest recording (standalone mode).

type Metrics ¶ added in v0.0.30

type Metrics struct {
	ActiveWorkers  int64
	QueuedWorkers  int64
	WorkerCount    int64
	ReservedBytes  int64
	BudgetBytes    int64
	PerArticleByte int64
}

Metrics is a point-in-time snapshot of engine activity for observability.

type Post ¶

type Post struct {
	FilePath string
	Articles []*article.Article
	Status   PostStatus
	Error    error
	Retries  int
	// contains filtered or unexported fields
}

Post represents a file to be posted

type PostStatus ¶

type PostStatus int

PostStatus represents the status of a post

const (
	PostStatusPending PostStatus = iota
	PostStatusPosted
	PostStatusVerified
	PostStatusFailed
	PostStatusCancelled
	PostStatusPosting
)

type Poster ¶

type Poster interface {
	// Post posts files from a directory to Usenet
	Post(ctx context.Context, files []string, rootDir string, nzbGen nzb.NZBGenerator) error
	// PostWithRelativePaths posts files with custom display names (relative paths) for subjects
	// relativePaths maps absolute file path to the display name to use in the subject
	PostWithRelativePaths(ctx context.Context, files []string, rootDir string, nzbGen nzb.NZBGenerator, relativePaths map[string]string) error
	// Stats returns posting statistics
	Stats() Stats
	// Close closes the poster
	Close()
}

Poster defines the interface for posting articles to Usenet

func New(ctx context.Context, cfg config.Config, poolManager pool.PoolManager, jobProgress progress.JobProgress) (Poster, error)

func NewWithEngine(ctx context.Context, cfg config.Config, poolManager pool.PoolManager, jobProgress progress.JobProgress, engine *Engine, manifestSink ManifestSink) (Poster, error)

type Reposter ¶ added in v0.0.30

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

Reposter re-posts individual articles during durable verification, reading each body from its original source file and posting through the shared upload pool and engine. It is process-wide (owned by the transfer runtime), distinct from the per-job poster, and implements verification.Reposter.

func NewReposter(uploadPool pool.NNTPClient, engine *Engine, throttleRate int64) *Reposter

func (r *Reposter) Repost(ctx context.Context, rec manifest.ArticleRecord) error

func (r *Reposter) Stats() Stats

type Stats ¶

type Stats struct {
	ArticlesPosted  int64
	ArticlesChecked int64
	BytesPosted     int64
	ArticleErrors   int64
	StartTime       time.Time
	// contains filtered or unexported fields
}

Stats tracks posting statistics

type Throttle ¶

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

Throttle handles rate limiting using a token-bucket algorithm. Token-bucket state is protected by a mutex: the previous lock-free implementation allowed concurrent consumers to overdraw the bucket past zero, silently bypassing the configured rate under contention.

func NewThrottle(rate int64, interval time.Duration) *Throttle

func (t *Throttle) Wait(bytes int64)