@@ -199,10 +208,15 @@ function formatNumber(num: number): string {
| {$t('metrics.server_host')} |
{$t('metrics.connections')} |
+ {$t('metrics.free_slots')} |
+ {$t('metrics.current_speed')} |
{$t('metrics.avg_speed')} |
+ {$t('metrics.downloaded')} |
{$t('metrics.missing')} |
+ {$t('metrics.ttfb')} |
{$t('metrics.ping_rtt')} |
{$t('metrics.errors')} |
+ {$t('metrics.quota')} |
@@ -216,16 +230,36 @@ function formatNumber(num: number): string {
{provider.activeConnections}/{provider.maxConnections}
+
+
+ {formatNumber(provider.availableSlots)}
+
+ |
+
+
+ {provider.speedEwma > 0 ? formatSpeed(provider.speedEwma) : "—"}
+
+ |
{formatSpeed(provider.avgSpeed)}
|
+
+
+ {provider.bytesConsumed > 0 ? formatBytes(provider.bytesConsumed) : "—"}
+
+ |
{formatNumber(provider.missing)}
|
+
+
+ {provider.ttfb || "—"}
+
+ |
{provider.pingRTT || "—"}
@@ -236,6 +270,18 @@ function formatNumber(num: number): string {
{formatNumber(provider.totalErrors)}
|
+
+ {#if provider.quotaBytes > 0}
+
+ {formatBytes(provider.quotaUsed)} / {formatBytes(provider.quotaBytes)}
+ {#if provider.quotaExceeded}
+ {$t('metrics.quota_exceeded')}
+ {/if}
+
+ {:else}
+ —
+ {/if}
+ |
{/each}
diff --git a/go.mod b/go.mod
index 5a1e83d..340b8c6 100644
--- a/go.mod
+++ b/go.mod
@@ -8,7 +8,7 @@ require (
github.com/google/uuid v1.6.0
github.com/gorilla/mux v1.8.1
github.com/gorilla/websocket v1.5.3
- github.com/javi11/nntppool/v4 v4.11.1
+ github.com/javi11/nntppool/v4 v4.13.0
github.com/javi11/nxg v0.1.0
github.com/javi11/nzbparser v0.5.4
github.com/javi11/par2go v0.0.10
diff --git a/go.sum b/go.sum
index d2894e0..1bf6ec1 100644
--- a/go.sum
+++ b/go.sum
@@ -345,8 +345,8 @@ github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo
github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
github.com/jackmordaunt/icns v1.0.0 h1:RYSxplerf/l/DUd09AHtITwckkv/mqjVv4DjYdPmAMQ=
github.com/jackmordaunt/icns v1.0.0/go.mod h1:7TTQVEuGzVVfOPPlLNHJIkzA6CoV7aH1Dv9dW351oOo=
-github.com/javi11/nntppool/v4 v4.11.1 h1:581fZSPv+RyIKY2hI0GB0eZrm1rCbAp+HRP5nIp7kd0=
-github.com/javi11/nntppool/v4 v4.11.1/go.mod h1:+UtisJLDFLXBSkW9R6uCRgdp3lS/+6pLAk7L+Wt6LMw=
+github.com/javi11/nntppool/v4 v4.13.0 h1:JEZukeSqeolKHw/M67YtVa2WrvZ8BE8UbkDklaU6/GE=
+github.com/javi11/nntppool/v4 v4.13.0/go.mod h1:+UtisJLDFLXBSkW9R6uCRgdp3lS/+6pLAk7L+Wt6LMw=
github.com/javi11/nxg v0.1.0 h1:CTThldYlaVIPIhpkrMw0HcTD0NLrW1uYMoDILjjEOtM=
github.com/javi11/nxg v0.1.0/go.mod h1:+GvYpp+y1oq+qBOWxFMvfTjtin/0zCeomWfjiPkiu8A=
github.com/javi11/nzbparser v0.5.4 h1:0aYyORZipp7iX8eNpT/efnzCeVO+9C0sE2HWCGc/JaI=
diff --git a/internal/backend/app.go b/internal/backend/app.go
index 7de6222..4470f76 100644
--- a/internal/backend/app.go
+++ b/internal/backend/app.go
@@ -81,6 +81,7 @@ type NntpPoolMetrics struct {
ActiveConnections int `json:"activeConnections"`
TotalErrors int64 `json:"totalErrors"`
AvgSpeed float64 `json:"avgSpeed"`
+ BytesConsumed int64 `json:"bytesConsumed"`
Elapsed string `json:"elapsed"`
ProviderErrors map[string]int64 `json:"providerErrors"`
Providers []NntpProviderMetrics `json:"providers"`
@@ -92,11 +93,19 @@ type NntpProviderMetrics struct {
Host string `json:"host"`
ActiveConnections int `json:"activeConnections"`
MaxConnections int `json:"maxConnections"`
+ AvailableSlots int `json:"availableSlots"`
TotalErrors int64 `json:"totalErrors"`
AvgSpeed float64 `json:"avgSpeed"`
+ SpeedEwma float64 `json:"speedEwma"` // recent throughput estimate (bytes/sec), 0 = no sample
+ BytesConsumed int64 `json:"bytesConsumed"`
Missing int64 `json:"missing"`
PingRTT string `json:"pingRTT"`
+ TTFB string `json:"ttfb"` // recent time-to-first-byte, "" = no sample
Inflight int `json:"inflight"`
+ QuotaBytes int64 `json:"quotaBytes"` // 0 = no quota configured
+ QuotaUsed int64 `json:"quotaUsed"`
+ QuotaResetAt string `json:"quotaResetAt"` // RFC3339, "" = no period
+ QuotaExceeded bool `json:"quotaExceeded"`
}
// App struct for the Wails application
@@ -1169,6 +1178,7 @@ func (a *App) GetNntpPoolMetrics() (NntpPoolMetrics, error) {
ActiveConnections: activeConnections,
TotalErrors: totalErrors,
AvgSpeed: stats.AvgSpeed,
+ BytesConsumed: stats.BytesConsumed,
Elapsed: stats.Elapsed.String(),
ProviderErrors: providerErrors,
}
@@ -1198,16 +1208,32 @@ func (a *App) GetNntpPoolMetrics() (NntpPoolMetrics, error) {
// Convert provider metrics from v4 ProviderStats
providers := make([]NntpProviderMetrics, 0, len(stats.Providers))
for _, provider := range stats.Providers {
+ ttfb := ""
+ if provider.TTFB > 0 {
+ ttfb = provider.TTFB.String()
+ }
+ quotaResetAt := ""
+ if !provider.QuotaResetAt.IsZero() {
+ quotaResetAt = provider.QuotaResetAt.Format(time.RFC3339)
+ }
providers = append(providers, NntpProviderMetrics{
Name: nameByAddr[provider.Name],
Host: provider.Name,
ActiveConnections: provider.ActiveConnections,
MaxConnections: provider.MaxConnections,
+ AvailableSlots: provider.AvailableSlots,
TotalErrors: provider.Errors,
AvgSpeed: provider.AvgSpeed,
+ SpeedEwma: provider.SpeedEWMA,
+ BytesConsumed: provider.BytesConsumed,
Missing: provider.Missing,
PingRTT: provider.Ping.RTT.String(),
+ TTFB: ttfb,
Inflight: inflightByAddr[provider.Name],
+ QuotaBytes: provider.QuotaBytes,
+ QuotaUsed: provider.QuotaUsed,
+ QuotaResetAt: quotaResetAt,
+ QuotaExceeded: provider.QuotaExceeded,
})
}
metrics.Providers = providers
diff --git a/internal/config/config.go b/internal/config/config.go
index aeedf64..37e5e58 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -271,6 +271,8 @@ type PostCheck struct {
DeferredCheckInterval Duration `yaml:"deferred_check_interval" json:"deferred_check_interval"`
// Number of articles processed per deferred check cycle. Default value is 500.
DeferredBatchSize int `yaml:"deferred_batch_size" json:"deferred_batch_size"`
+ // Number of segments checked per batched STAT sweep. Default value is 100.
+ StatBatchSize int `yaml:"stat_batch_size" json:"stat_batch_size"`
// MaxConcurrentChecks caps the number of concurrent STAT verification checks
// across the whole process. A value of 0 enables automatic sizing: the
// durable verification service uses a dedicated verification pool capped at
@@ -582,6 +584,9 @@ func Load(path string) (*ConfigData, error) {
if cfg.PostCheck.DeferredBatchSize <= 0 {
cfg.PostCheck.DeferredBatchSize = 500
}
+ if cfg.PostCheck.StatBatchSize <= 0 {
+ cfg.PostCheck.StatBatchSize = 100
+ }
if cfg.Par2.Redundancy == "" {
cfg.Par2.Redundancy = defaultRedundancy
@@ -1065,6 +1070,8 @@ func GetDefaultConfig() ConfigData {
DeferredMaxRetries: 5,
DeferredMaxBackoff: Duration("1h"),
DeferredCheckInterval: Duration("2m"),
+ DeferredBatchSize: 500,
+ StatBatchSize: 100,
},
Par2: Par2Config{
Enabled: &enabled,
diff --git a/internal/mocks/nntpclient.go b/internal/mocks/nntpclient.go
index ba74a0c..0c8c000 100644
--- a/internal/mocks/nntpclient.go
+++ b/internal/mocks/nntpclient.go
@@ -65,6 +65,20 @@ func (mr *MockNNTPClientMockRecorder) Stat(ctx, messageID any) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Stat", reflect.TypeOf((*MockNNTPClient)(nil).Stat), ctx, messageID)
}
+// StatMany mocks base method.
+func (m *MockNNTPClient) StatMany(ctx context.Context, messageIDs []string, opts nntppool.StatManyOptions) <-chan nntppool.StatManyResult {
+ m.ctrl.T.Helper()
+ ret := m.ctrl.Call(m, "StatMany", ctx, messageIDs, opts)
+ ret0, _ := ret[0].(<-chan nntppool.StatManyResult)
+ return ret0
+}
+
+// StatMany indicates an expected call of StatMany.
+func (mr *MockNNTPClientMockRecorder) StatMany(ctx, messageIDs, opts any) *gomock.Call {
+ mr.mock.ctrl.T.Helper()
+ return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StatMany", reflect.TypeOf((*MockNNTPClient)(nil).StatMany), ctx, messageIDs, opts)
+}
+
// Stats mocks base method.
func (m *MockNNTPClient) Stats() nntppool.ClientStats {
m.ctrl.T.Helper()
diff --git a/internal/pool/nntpclient.go b/internal/pool/nntpclient.go
index ad154e4..fd6c50f 100644
--- a/internal/pool/nntpclient.go
+++ b/internal/pool/nntpclient.go
@@ -13,8 +13,53 @@ import (
type NNTPClient interface {
PostYenc(ctx context.Context, headers nntppool.PostHeaders, body io.Reader, meta rapidyenc.Meta) (*nntppool.PostResult, error)
Stat(ctx context.Context, messageID string) (*nntppool.StatResult, error)
+ StatMany(ctx context.Context, messageIDs []string, opts nntppool.StatManyOptions) <-chan nntppool.StatManyResult
Stats() nntppool.ClientStats
AddProvider(p nntppool.Provider) error
RemoveProvider(name string) error
Close() error
}
+
+// DefaultStatBatchSize is the fallback number of message-IDs per StatMany call
+// when no batch size is configured.
+const DefaultStatBatchSize = 100
+
+// StatMissing STATs ids in chunks of batchSize (<=0 -> DefaultStatBatchSize)
+// and returns the message-IDs NOT confirmed present. Any per-ID error counts
+// as missing (preserves the "error => repost" semantics of the previous
+// one-by-one checks). Returns ctx.Err() if the sweep was cancelled.
+func StatMissing(ctx context.Context, c NNTPClient, ids []string, batchSize int) (map[string]struct{}, error) {
+ missing := make(map[string]struct{})
+ if len(ids) == 0 {
+ return missing, nil
+ }
+ if batchSize <= 0 {
+ batchSize = DefaultStatBatchSize
+ }
+
+ for start := 0; start < len(ids); start += batchSize {
+ chunk := ids[start:min(start+batchSize, len(ids))]
+
+ reported := make(map[string]struct{}, len(chunk))
+ for res := range c.StatMany(ctx, chunk, nntppool.StatManyOptions{}) {
+ reported[res.MessageID] = struct{}{}
+ if res.Err != nil {
+ missing[res.MessageID] = struct{}{}
+ }
+ }
+ if err := ctx.Err(); err != nil {
+ return missing, err
+ }
+ // An interrupted sweep may drop results; ids that never reported must
+ // count as missing rather than silently pass as present.
+ if len(reported) != len(chunk) {
+ for _, id := range chunk {
+ if _, ok := reported[id]; !ok {
+ missing[id] = struct{}{}
+ }
+ }
+ }
+ }
+
+ return missing, nil
+}
diff --git a/internal/pool/nntpclient_test.go b/internal/pool/nntpclient_test.go
new file mode 100644
index 0000000..a44719a
--- /dev/null
+++ b/internal/pool/nntpclient_test.go
@@ -0,0 +1,156 @@
+package pool_test
+
+import (
+ "context"
+ "errors"
+ "testing"
+
+ "github.com/javi11/nntppool/v4"
+ "go.uber.org/mock/gomock"
+
+ "github.com/javi11/postie/internal/mocks"
+ "github.com/javi11/postie/internal/pool"
+)
+
+// statManyStub answers each id as present unless it has an error in errs. It
+// also records the chunk sizes it was called with via calls.
+func statManyStub(errs map[string]error, calls *[][]string) func(context.Context, []string, nntppool.StatManyOptions) <-chan nntppool.StatManyResult {
+ return func(_ context.Context, ids []string, _ nntppool.StatManyOptions) <-chan nntppool.StatManyResult {
+ if calls != nil {
+ *calls = append(*calls, ids)
+ }
+ out := make(chan nntppool.StatManyResult, len(ids))
+ for _, id := range ids {
+ res := nntppool.StatManyResult{MessageID: id, Result: &nntppool.StatResult{MessageID: id}}
+ if err, ok := errs[id]; ok {
+ res = nntppool.StatManyResult{MessageID: id, Err: err}
+ }
+ out <- res
+ }
+ close(out)
+ return out
+ }
+}
+
+func TestStatMissing_EmptyInput(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ client := mocks.NewMockNNTPClient(ctrl) // no StatMany expected
+
+ missing, err := pool.StatMissing(context.Background(), client, nil, 100)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if len(missing) != 0 {
+ t.Errorf("expected no missing ids, got %v", missing)
+ }
+}
+
+func TestStatMissing_MissesReported(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ client := mocks.NewMockNNTPClient(ctrl)
+ errs := map[string]error{
+ "m1": nntppool.ErrArticleNotFound,
+ "m3": errors.New("connection died"),
+ }
+ client.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).
+ DoAndReturn(statManyStub(errs, nil)).Times(1)
+
+ missing, err := pool.StatMissing(context.Background(), client, []string{"m0", "m1", "m2", "m3"}, 100)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ // Both a genuine 430 miss and a generic error count as missing.
+ for _, id := range []string{"m1", "m3"} {
+ if _, ok := missing[id]; !ok {
+ t.Errorf("expected %s to be missing", id)
+ }
+ }
+ for _, id := range []string{"m0", "m2"} {
+ if _, ok := missing[id]; ok {
+ t.Errorf("expected %s to be present", id)
+ }
+ }
+}
+
+func TestStatMissing_ChunksByBatchSize(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ client := mocks.NewMockNNTPClient(ctrl)
+
+ var calls [][]string
+ client.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).
+ DoAndReturn(statManyStub(nil, &calls)).Times(3)
+
+ ids := []string{"a", "b", "c", "d", "e"}
+ if _, err := pool.StatMissing(context.Background(), client, ids, 2); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ want := [][]int{{2}, {2}, {1}}
+ if len(calls) != 3 || len(calls[0]) != 2 || len(calls[1]) != 2 || len(calls[2]) != 1 {
+ t.Errorf("expected chunk sizes 2,2,1 (%v), got %v", want, calls)
+ }
+}
+
+func TestStatMissing_UnreportedIDsCountAsMissing(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ client := mocks.NewMockNNTPClient(ctrl)
+
+ // Simulate an interrupted sweep that drops the second id entirely.
+ client.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+ func(_ context.Context, ids []string, _ nntppool.StatManyOptions) <-chan nntppool.StatManyResult {
+ out := make(chan nntppool.StatManyResult, 1)
+ out <- nntppool.StatManyResult{MessageID: ids[0], Result: &nntppool.StatResult{MessageID: ids[0]}}
+ close(out)
+ return out
+ }).Times(1)
+
+ missing, err := pool.StatMissing(context.Background(), client, []string{"a", "b"}, 100)
+ if err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if _, ok := missing["b"]; !ok {
+ t.Error("expected unreported id b to count as missing")
+ }
+ if _, ok := missing["a"]; ok {
+ t.Error("expected reported id a to be present")
+ }
+}
+
+func TestStatMissing_CancelledContextReturnsError(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ client := mocks.NewMockNNTPClient(ctrl)
+ client.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).
+ DoAndReturn(statManyStub(nil, nil)).AnyTimes()
+
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+
+ if _, err := pool.StatMissing(ctx, client, []string{"a"}, 100); !errors.Is(err, context.Canceled) {
+ t.Errorf("expected context.Canceled, got %v", err)
+ }
+}
+
+func TestStatMissing_ZeroBatchSizeUsesDefault(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+ client := mocks.NewMockNNTPClient(ctrl)
+
+ var calls [][]string
+ client.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).
+ DoAndReturn(statManyStub(nil, &calls)).Times(2)
+
+ ids := make([]string, pool.DefaultStatBatchSize+1)
+ for i := range ids {
+ ids[i] = string(rune('a' + i%26))
+ }
+ if _, err := pool.StatMissing(context.Background(), client, ids, 0); err != nil {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ if len(calls) != 2 || len(calls[0]) != pool.DefaultStatBatchSize || len(calls[1]) != 1 {
+ t.Errorf("expected chunks of %d and 1, got sizes %d", pool.DefaultStatBatchSize, len(calls))
+ }
+}
diff --git a/internal/poster/poster.go b/internal/poster/poster.go
index b87499e..39a3aed 100644
--- a/internal/poster/poster.go
+++ b/internal/poster/poster.go
@@ -1113,34 +1113,28 @@ func (p *poster) addRecoveredPost(ctx context.Context, filePath string, file *os
}
}
-// filterMissing STATs each article (bounded concurrency) and returns those that
-// are NOT already present on the server, preserving order. On a STAT error the
-// article is treated as missing (re-posting an already-present article is
-// idempotent). If no verify pool is available, all articles are returned.
+// filterMissing STATs the articles in batches and returns those that are NOT
+// already present on the server, preserving order. On a STAT error the article
+// is treated as missing (re-posting an already-present article is idempotent).
+// If no verify pool is available, all articles are returned.
func (p *poster) filterMissing(ctx context.Context, articles []*article.Article) []*article.Article {
if p.verifyPool == nil || len(articles) == 0 {
return articles
}
- keep := make([]bool, len(articles))
- sem := make(chan struct{}, 16)
- var wg sync.WaitGroup
+ ids := make([]string, len(articles))
for i, art := range articles {
- sem <- struct{}{}
- wg.Add(1)
- go func() {
- defer wg.Done()
- defer func() { <-sem }()
- if _, err := p.verifyPool.Stat(ctx, art.MessageID); err != nil {
- keep[i] = true // missing (or unknown) -> re-post
- }
- }()
+ ids[i] = art.MessageID
+ }
+
+ missingIDs, err := pool.StatMissing(ctx, p.verifyPool, ids, p.checkCfg.StatBatchSize)
+ if err != nil {
+ return articles // cancelled sweep: fall back to re-posting everything
}
- wg.Wait()
missing := make([]*article.Article, 0, len(articles))
- for i, art := range articles {
- if keep[i] {
+ for _, art := range articles {
+ if _, ok := missingIDs[art.MessageID]; ok {
missing = append(missing, art)
}
}
diff --git a/internal/poster/recovery_test.go b/internal/poster/recovery_test.go
index 01871da..d577af4 100644
--- a/internal/poster/recovery_test.go
+++ b/internal/poster/recovery_test.go
@@ -18,9 +18,9 @@ func TestFilterMissing_KeepsOnlyMissingPreservingOrder(t *testing.T) {
mockPool := mocks.NewMockNNTPClient(ctrl)
// m0 and m2 already present; m1 missing.
- mockPool.EXPECT().Stat(gomock.Any(), "m0").Return(&nntppool.StatResult{}, nil).AnyTimes()
- mockPool.EXPECT().Stat(gomock.Any(), "m1").Return(nil, errors.New("430 no such article")).AnyTimes()
- mockPool.EXPECT().Stat(gomock.Any(), "m2").Return(&nntppool.StatResult{}, nil).AnyTimes()
+ mockPool.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+ statManyStub(map[string]error{"m1": errors.New("430 no such article")}),
+ ).AnyTimes()
p := &poster{verifyPool: mockPool}
got := p.filterMissing(context.Background(), []*article.Article{
@@ -36,7 +36,9 @@ func TestFilterMissing_AllPresentReturnsEmpty(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
mockPool := mocks.NewMockNNTPClient(ctrl)
- mockPool.EXPECT().Stat(gomock.Any(), gomock.Any()).Return(&nntppool.StatResult{}, nil).AnyTimes()
+ mockPool.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
+ statManyStub(nil),
+ ).AnyTimes()
p := &poster{verifyPool: mockPool}
got := p.filterMissing(context.Background(), []*article.Article{{MessageID: "a"}, {MessageID: "b"}})
@@ -54,6 +56,23 @@ func TestFilterMissing_NoVerifyPoolRepostsAll(t *testing.T) {
}
}
+// statManyStub builds a StatMany stub that reports each id present unless it
+// has an error in errs.
+func statManyStub(errs map[string]error) func(context.Context, []string, nntppool.StatManyOptions) <-chan nntppool.StatManyResult {
+ return func(_ context.Context, ids []string, _ nntppool.StatManyOptions) <-chan nntppool.StatManyResult {
+ out := make(chan nntppool.StatManyResult, len(ids))
+ for _, id := range ids {
+ res := nntppool.StatManyResult{MessageID: id, Result: &nntppool.StatResult{MessageID: id}}
+ if err, ok := errs[id]; ok {
+ res = nntppool.StatManyResult{MessageID: id, Err: err}
+ }
+ out <- res
+ }
+ close(out)
+ return out
+ }
+}
+
func msgIDs(arts []*article.Article) []string {
ids := make([]string, len(arts))
for i, a := range arts {
diff --git a/internal/processor/postcheck_retry_worker.go b/internal/processor/postcheck_retry_worker.go
index dec96ec..5014557 100644
--- a/internal/processor/postcheck_retry_worker.go
+++ b/internal/processor/postcheck_retry_worker.go
@@ -37,6 +37,7 @@ type PostCheckRetryWorker struct {
maxBackoff time.Duration
maxRetries int
batchSize int
+ statBatchSize int
onStatusChanged func() // notified when a completed item's verification status is updated
}
@@ -75,6 +76,11 @@ func NewPostCheckRetryWorker(
batchSize = 500
}
+ statBatchSize := cfg.StatBatchSize
+ if statBatchSize <= 0 {
+ statBatchSize = pool.DefaultStatBatchSize
+ }
+
return &PostCheckRetryWorker{
queue: q,
checkPool: checkPool,
@@ -86,6 +92,7 @@ func NewPostCheckRetryWorker(
maxBackoff: maxBackoff,
maxRetries: maxRetries,
batchSize: batchSize,
+ statBatchSize: statBatchSize,
onStatusChanged: onStatusChanged,
}
}
@@ -153,14 +160,11 @@ func (w *PostCheckRetryWorker) processRetries() bool {
// Track completed items that need status updates
completedItems := make(map[string]bool)
+ // Parse groups up front so malformed rows are failed before the STAT sweep.
+ checkable := make([]queue.PendingArticleCheck, 0, len(articles))
for _, article := range articles {
- if ctx.Err() != nil {
- return false
- }
-
completedItems[article.CompletedItemID] = true
- // Parse groups from JSON
var groups []string
if err := json.Unmarshal([]byte(article.Groups), &groups); err != nil {
slog.ErrorContext(ctx, "Failed to parse groups JSON", "error", err, "articleID", article.ID)
@@ -169,11 +173,28 @@ func (w *PostCheckRetryWorker) processRetries() bool {
}
continue
}
+ checkable = append(checkable, article)
+ }
+
+ // Run batched STAT checks over all remaining articles.
+ ids := make([]string, len(checkable))
+ for i, article := range checkable {
+ ids[i] = article.MessageID
+ }
+ missing, err := pool.StatMissing(ctx, w.checkPool, ids, w.statBatchSize)
+ if err != nil {
+ slog.WarnContext(ctx, "Deferred STAT sweep interrupted", "error", err)
+ return false
+ }
+
+ for _, article := range checkable {
+ if ctx.Err() != nil {
+ return false
+ }
- // Run STAT check
- verified := w.checkArticle(ctx, article.MessageID, groups)
+ _, isMissing := missing[article.MessageID]
- if verified {
+ if !isMissing {
if err := w.queue.MarkArticleVerified(ctx, article.ID); err != nil {
slog.ErrorContext(ctx, "Failed to mark article as verified", "error", err, "articleID", article.ID)
} else {
@@ -220,13 +241,6 @@ func (w *PostCheckRetryWorker) processRetries() bool {
return len(articles) == w.batchSize
}
-// checkArticle verifies if an article exists on the server via STAT command
-func (w *PostCheckRetryWorker) checkArticle(ctx context.Context, messageID string, groups []string) bool {
- // v4 Stat only takes messageID (no groups parameter)
- _, err := w.checkPool.Stat(ctx, messageID)
- return err == nil
-}
-
// calculateBackoff calculates the exponential backoff delay
func (w *PostCheckRetryWorker) calculateBackoff(retryCount int) time.Duration {
// Exponential backoff: initialDelay * 2^retryCount
diff --git a/internal/processor/postcheck_retry_worker_test.go b/internal/processor/postcheck_retry_worker_test.go
index fef5b56..7d4d05f 100644
--- a/internal/processor/postcheck_retry_worker_test.go
+++ b/internal/processor/postcheck_retry_worker_test.go
@@ -7,6 +7,7 @@ import (
"testing"
"time"
+ "github.com/javi11/nntppool/v4"
"github.com/javi11/postie/internal/config"
"github.com/javi11/postie/internal/mocks"
"github.com/javi11/postie/internal/queue"
@@ -123,7 +124,7 @@ func TestProcessRetries(t *testing.T) {
articles := makeArticles(2, 0)
q := &fakeQueue{articles: articles, countPend: 1}
mockPool := mocks.NewMockNNTPClient(ctrl)
- mockPool.EXPECT().Stat(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
+ mockPool.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(statManyStub(nil)).AnyTimes()
w := newWorker(context.Background(), q, mockPool, 3, 5)
got := w.processRetries()
@@ -139,7 +140,7 @@ func TestProcessRetries(t *testing.T) {
articles := makeArticles(2, 0)
q := &fakeQueue{articles: articles, countTotal: 2, countPend: 0}
mockPool := mocks.NewMockNNTPClient(ctrl)
- mockPool.EXPECT().Stat(gomock.Any(), gomock.Any()).Return(nil, nil).Times(2)
+ mockPool.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(statManyStub(nil)).Times(1)
w := newWorker(context.Background(), q, mockPool, 2, 5)
got := w.processRetries()
@@ -155,7 +156,7 @@ func TestProcessRetries(t *testing.T) {
articles := makeArticles(1, 0)
q := &fakeQueue{articles: articles, countTotal: 1, countPend: 0}
mockPool := mocks.NewMockNNTPClient(ctrl)
- mockPool.EXPECT().Stat(gomock.Any(), articles[0].MessageID).Return(nil, nil).Times(1)
+ mockPool.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(statManyStub(nil)).Times(1)
w := newWorker(context.Background(), q, mockPool, 10, 5)
w.processRetries()
@@ -175,7 +176,7 @@ func TestProcessRetries(t *testing.T) {
articles := makeArticles(1, 0) // retryCount=0, maxRetries=3 → newRetryCount=1 < 3
q := &fakeQueue{articles: articles, countTotal: 1, countPend: 1}
mockPool := mocks.NewMockNNTPClient(ctrl)
- mockPool.EXPECT().Stat(gomock.Any(), articles[0].MessageID).Return(nil, errors.New("not found")).Times(1)
+ mockPool.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(statManyStub(map[string]error{articles[0].MessageID: errors.New("not found")})).Times(1)
w := newWorker(context.Background(), q, mockPool, 10, 3)
w.processRetries()
@@ -196,7 +197,7 @@ func TestProcessRetries(t *testing.T) {
articles := makeArticles(1, 2)
q := &fakeQueue{articles: articles, countTotal: 1, countPend: 0, countFailed: 1}
mockPool := mocks.NewMockNNTPClient(ctrl)
- mockPool.EXPECT().Stat(gomock.Any(), articles[0].MessageID).Return(nil, errors.New("not found")).Times(1)
+ mockPool.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(statManyStub(map[string]error{articles[0].MessageID: errors.New("not found")})).Times(1)
w := newWorker(context.Background(), q, mockPool, 10, 3)
w.processRetries()
@@ -220,8 +221,8 @@ func TestProcessRetries(t *testing.T) {
q := &fakeQueue{articles: articles}
mockPool := mocks.NewMockNNTPClient(ctrl)
// With cancelled ctx, the worker should detect ctx.Err() before processing articles
- // Stat may or may not be called depending on timing, so allow any calls
- mockPool.EXPECT().Stat(gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
+ // StatMany may or may not be called depending on timing, so allow any calls
+ mockPool.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(statManyStub(nil)).AnyTimes()
w := newWorker(ctx, q, mockPool, 2, 5)
got := w.processRetries()
@@ -251,7 +252,7 @@ func TestProcessRetries(t *testing.T) {
articles := makeArticles(1, 0)
q := &fakeQueue{articles: articles, countTotal: 1, countPend: 0, countFailed: 0}
mockPool := mocks.NewMockNNTPClient(ctrl)
- mockPool.EXPECT().Stat(gomock.Any(), articles[0].MessageID).Return(nil, nil).Times(1)
+ mockPool.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(statManyStub(nil)).Times(1)
called := false
enabled := makeEnabled(true)
@@ -283,7 +284,7 @@ func TestProcessRetries(t *testing.T) {
articles := makeArticles(1, 4)
q := &fakeQueue{articles: articles, countTotal: 1, countPend: 0, countFailed: 1}
mockPool := mocks.NewMockNNTPClient(ctrl)
- mockPool.EXPECT().Stat(gomock.Any(), articles[0].MessageID).Return(nil, errors.New("not found")).Times(1)
+ mockPool.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(statManyStub(map[string]error{articles[0].MessageID: errors.New("not found")})).Times(1)
called := false
enabled := makeEnabled(true)
@@ -311,10 +312,10 @@ func TestProcessRetries(t *testing.T) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
- articles := makeArticles(1, 0) // retryCount=0, maxRetries=5 → schedules retry
+ articles := makeArticles(1, 0) // retryCount=0, maxRetries=5 → schedules retry
q := &fakeQueue{articles: articles, countTotal: 1, countPend: 1} // still pending
mockPool := mocks.NewMockNNTPClient(ctrl)
- mockPool.EXPECT().Stat(gomock.Any(), articles[0].MessageID).Return(nil, errors.New("not found")).Times(1)
+ mockPool.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(statManyStub(map[string]error{articles[0].MessageID: errors.New("not found")})).Times(1)
called := false
w := newWorker(context.Background(), q, mockPool, 10, 5)
@@ -343,7 +344,7 @@ func TestProcessRetries(t *testing.T) {
statusErr: errors.New("db error"),
}
mockPool := mocks.NewMockNNTPClient(ctrl)
- mockPool.EXPECT().Stat(gomock.Any(), articles[0].MessageID).Return(nil, nil).Times(1)
+ mockPool.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(statManyStub(nil)).Times(1)
called := false
w := newWorker(context.Background(), q, mockPool, 10, 5)
@@ -363,7 +364,7 @@ func TestProcessRetries(t *testing.T) {
articles := makeArticles(1, 0)
q := &fakeQueue{articles: articles, countTotal: 1, countPend: 0}
mockPool := mocks.NewMockNNTPClient(ctrl)
- mockPool.EXPECT().Stat(gomock.Any(), articles[0].MessageID).Return(nil, nil).Times(1)
+ mockPool.EXPECT().StatMany(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(statManyStub(nil)).Times(1)
// nil callback — should not panic
w := newWorker(context.Background(), q, mockPool, 10, 5)
@@ -400,3 +401,20 @@ func TestProcessRetries(t *testing.T) {
}
})
}
+
+// statManyStub builds a StatMany stub that reports each id present unless it
+// has an error in errs.
+func statManyStub(errs map[string]error) func(context.Context, []string, nntppool.StatManyOptions) <-chan nntppool.StatManyResult {
+ return func(_ context.Context, ids []string, _ nntppool.StatManyOptions) <-chan nntppool.StatManyResult {
+ out := make(chan nntppool.StatManyResult, len(ids))
+ for _, id := range ids {
+ res := nntppool.StatManyResult{MessageID: id, Result: &nntppool.StatResult{MessageID: id}}
+ if err, ok := errs[id]; ok {
+ res = nntppool.StatManyResult{MessageID: id, Err: err}
+ }
+ out <- res
+ }
+ close(out)
+ return out
+ }
+}
diff --git a/internal/verification/service.go b/internal/verification/service.go
index 9cb3717..a946f34 100644
--- a/internal/verification/service.go
+++ b/internal/verification/service.go
@@ -19,17 +19,19 @@ import (
"errors"
"io"
"log/slog"
- "sync"
"time"
"github.com/javi11/postie/internal/manifest"
"github.com/javi11/postie/internal/transferstore"
)
-// Stater reports whether an article exists on the server. A nil error means the
-// article was found; any error means it is (still) missing.
+// Stater reports whether articles exist on the server. For Stat, a nil error
+// means the article was found; any error means it is (still) missing. For
+// StatBatch, the returned set contains the message-IDs NOT confirmed present
+// (any per-ID error counts as missing).
type Stater interface {
Stat(ctx context.Context, messageID string) error
+ StatBatch(ctx context.Context, messageIDs []string) (missing map[string]struct{}, err error)
}
// Reposter re-posts a single article described by its manifest record, reusing
@@ -51,6 +53,9 @@ type Cleaner interface {
type Config struct {
// MaxConcurrentChecks bounds simultaneous STAT requests. 0 = auto (16).
MaxConcurrentChecks int
+ // StatBatchSize is the number of message-IDs checked per StatBatch call
+ // during file verification. 0 = 100.
+ StatBatchSize int
// PropagationDelay is how long to wait after a (re)post before checking.
PropagationDelay time.Duration
// MaxReposts is the maximum number of times a missing article is re-posted
@@ -76,6 +81,9 @@ func (c Config) withDefaults() Config {
if c.MaxConcurrentChecks <= 0 {
c.MaxConcurrentChecks = 16
}
+ if c.StatBatchSize <= 0 {
+ c.StatBatchSize = 100
+ }
if c.PropagationDelay <= 0 {
c.PropagationDelay = 10 * time.Second
}
@@ -251,43 +259,54 @@ func (s *Service) VerifyFile(ctx context.Context, tf transferstore.TransferFile)
}
defer func() { _ = r.Close() }()
- sem := make(chan struct{}, s.cfg.MaxConcurrentChecks)
- var (
- wg sync.WaitGroup
- mu sync.Mutex
- missing []manifest.ArticleRecord
- statErrs int
- )
+ var missing []manifest.ArticleRecord
+
+ // flush STATs a chunk of records in one batched sweep and collects misses.
+ flush := func(chunk []manifest.ArticleRecord) error {
+ if len(chunk) == 0 {
+ return nil
+ }
+ ids := make([]string, len(chunk))
+ for i, rec := range chunk {
+ ids[i] = rec.MessageID
+ }
+ missingIDs, err := s.stater.StatBatch(ctx, ids)
+ if err != nil {
+ return err
+ }
+ for _, rec := range chunk {
+ if _, ok := missingIDs[rec.MessageID]; ok {
+ missing = append(missing, rec)
+ }
+ }
+ return nil
+ }
+ chunk := make([]manifest.ArticleRecord, 0, s.cfg.StatBatchSize)
for {
rec, err := r.Next()
if errors.Is(err, io.EOF) {
break
}
if err != nil {
- wg.Wait()
return err
}
if err := ctx.Err(); err != nil {
- wg.Wait()
return err
}
- sem <- struct{}{}
- wg.Add(1)
- go func(rec manifest.ArticleRecord) {
- defer wg.Done()
- defer func() { <-sem }()
- if statErr := s.stater.Stat(ctx, rec.MessageID); statErr != nil {
- mu.Lock()
- missing = append(missing, rec)
- statErrs++
- mu.Unlock()
+ chunk = append(chunk, rec)
+ if len(chunk) >= s.cfg.StatBatchSize {
+ if err := flush(chunk); err != nil {
+ return err
}
- }(rec)
+ chunk = chunk[:0]
+ }
+ }
+ if err := flush(chunk); err != nil {
+ return err
}
- wg.Wait()
if err := ctx.Err(); err != nil {
return err
diff --git a/internal/verification/service_test.go b/internal/verification/service_test.go
index 97ecce8..68396a3 100644
--- a/internal/verification/service_test.go
+++ b/internal/verification/service_test.go
@@ -40,6 +40,16 @@ func (f *fakeStater) Stat(_ context.Context, messageID string) error {
return nil
}
+func (f *fakeStater) StatBatch(ctx context.Context, messageIDs []string) (map[string]struct{}, error) {
+ missing := make(map[string]struct{})
+ for _, id := range messageIDs {
+ if err := f.Stat(ctx, id); err != nil {
+ missing[id] = struct{}{}
+ }
+ }
+ return missing, nil
+}
+
func (f *fakeStater) markPresent(id string) {
f.mu.Lock()
defer f.mu.Unlock()
diff --git a/pkg/postie/runtime.go b/pkg/postie/runtime.go
index d78b9be..586bd75 100644
--- a/pkg/postie/runtime.go
+++ b/pkg/postie/runtime.go
@@ -25,6 +25,12 @@ func (s poolStater) Stat(ctx context.Context, messageID string) error {
return err
}
+// StatBatch checks the ids in one batched sweep; the verification service
+// pre-chunks to its configured StatBatchSize, so the whole slice is one chunk.
+func (s poolStater) StatBatch(ctx context.Context, messageIDs []string) (map[string]struct{}, error) {
+ return pool.StatMissing(ctx, s.pool, messageIDs, len(messageIDs))
+}
+
// newPostVerifyScriptRunner returns a transfercleaner.ScriptRunner that runs
// the post-upload script once a transfer is verified, resolving the NZB path
// from the completed item and the source path from the transfer's files. Returns
@@ -59,6 +65,7 @@ func newPostVerifyScriptRunner(store *transferstore.Store, cfg config.PostUpload
func verificationConfig(pc config.PostCheck) verification.Config {
return verification.Config{
MaxConcurrentChecks: pc.MaxConcurrentChecks,
+ StatBatchSize: pc.StatBatchSize,
PropagationDelay: pc.RetryDelay.ToDuration(),
MaxReposts: int(pc.MaxRePost),
DeferredBackoff: pc.DeferredCheckDelay.ToDuration(),