youfak/proxy-pool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add activity health and inventory contracts

Browse Source
This commit is contained in:
youfak 2026-07-29 15:03:16 +08:00
parent c8e64758da
commit 3da836aeff
7 changed files with 397 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
internal/controller/extraction/service_test.go
View File

@ -153,6 +153,7 @@ func TestServiceIdempotentReplayKeepsOriginalExtractionTime(t *testing.T) {
upserted, err := store.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
ObservedAt: firstTime,
ConfiguredTTL: 2 * time.Minute,
MaxSize: 100,
Proxies: []proxyDomain.Proxy{{
ID: "p1", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080,
State: proxyDomain.StateAvailable, LastCheckedAt: &checkedAt,

2
internal/controller/pool/ownership_test.go
View File

@ -301,7 +301,7 @@ func newTestActivityPool(t *testing.T, now time.Time, proxyIDs ...string) *activ
}
store := activitypool.NewMemoryPool()
result, err := store.UpsertFetched(context.Background(), "provider-a", activitypool.FetchedBatch{
ObservedAt: now, ConfiguredTTL: 10 * time.Minute, Proxies: proxies,
ObservedAt: now, ConfiguredTTL: 10 * time.Minute, MaxSize: 100, Proxies: proxies,
})
if err != nil || result.Inserted != len(proxyIDs) {
t.Fatalf("UpsertFetched() = %+v, %v", result, err)

5
internal/controller/provider/reconciler.go
View File

@ -19,6 +19,7 @@ type Config struct {
Timeout time.Duration
MaxAttempts int
MaxInFlight int
MaxSize int
TTL time.Duration
AllocationSafetyMargin time.Duration
Retry RetryConfig
@ -62,7 +63,8 @@ func NewReconciler(config Config, ports Ports, runtimes ...Runtime) (*Reconciler
if config.UpstreamID == "" {
return nil, fmt.Errorf("new provider reconciler: upstream ID is required")
}
if config.RequestInterval < 0 || config.Timeout <= 0 || config.MaxAttempts <= 0 || config.MaxInFlight <= 0 {
if config.RequestInterval < 0 || config.Timeout <= 0 || config.MaxAttempts <= 0 ||
config.MaxInFlight <= 0 || config.MaxSize <= 0 {
return nil, fmt.Errorf("new provider reconciler: fetch limits must be positive")
}
if config.TTL < 0 || config.AllocationSafetyMargin < 0 ||
@ -213,6 +215,7 @@ func (r *Reconciler) fetchAttempt(ctx context.Context, attempt int) (FetchRespon
ObservedAt: r.runtime.Clock.Now().UTC(),
ConfiguredTTL: r.config.TTL,
AllocationSafetyMargin: r.config.AllocationSafetyMargin,
MaxSize: r.config.MaxSize,
Proxies: retained,
})
candidateErr = err

37
internal/controller/provider/reconciler_test.go
View File

@ -31,6 +31,7 @@ func TestReconcilerWritesProviderTTLPolicyToEphemeralPool(t *testing.T) {
Timeout: time.Second,
MaxAttempts: 1,
MaxInFlight: 1,
MaxSize: 321,
TTL: 30 * time.Second,
AllocationSafetyMargin: 3 * time.Second,
}, ports, Runtime{Clock: clock})
@ -44,7 +45,7 @@ func TestReconcilerWritesProviderTTLPolicyToEphemeralPool(t *testing.T) {
}
batch := <-batches
if !batch.ObservedAt.Equal(now) || batch.ConfiguredTTL != 30*time.Second ||
batch.AllocationSafetyMargin != 3*time.Second {
batch.AllocationSafetyMargin != 3*time.Second || batch.MaxSize != 321 {
t.Fatalf("activity batch = %+v", batch)
}
}
@ -57,6 +58,7 @@ func TestReconcilerCoalescesConcurrentNotifications(t *testing.T) {
Timeout: time.Second,
MaxAttempts: 1,
MaxInFlight: 1,
MaxSize: 100,
}, Ports{
Adapter: adapterFunc(func(context.Context) (FetchResponse, error) {
calls.Add(1)
@ -117,6 +119,7 @@ func TestReconcilerEnforcesRequestInterval(t *testing.T) {
Timeout: time.Second,
MaxAttempts: 1,
MaxInFlight: 1,
MaxSize: 100,
}, successfulPorts(func() { calledAt <- clock.Now() }, results), Runtime{
Clock: clock,
Sleeper: sleeper,
@ -167,6 +170,7 @@ func TestReconcilerRetriesErrorsWithExponentialBackoffAndJitter(t *testing.T) {
Timeout: time.Second,
MaxAttempts: 3,
MaxInFlight: 1,
MaxSize: 100,
Retry: RetryConfig{
Initial: 100 * time.Millisecond,
Max: time.Second,
@ -244,6 +248,7 @@ func TestReconcilerClassifiesFetchResults(t *testing.T) {
Timeout: time.Second,
MaxAttempts: 1,
MaxInFlight: 1,
MaxSize: 100,
}, ports)
if err != nil {
t.Fatalf("NewReconciler(): %v", err)
@ -277,6 +282,7 @@ func TestReconcilerHonorsRetryAfterBeforeBackoff(t *testing.T) {
Timeout: time.Second,
MaxAttempts: 2,
MaxInFlight: 1,
MaxSize: 100,
Retry: RetryConfig{Initial: 50 * time.Millisecond, Max: time.Second, Jitter: 20},
}, ports, Runtime{Clock: clock, Sleeper: sleeper, Random: fixedRandom(1)})
if err != nil {
@ -317,6 +323,7 @@ func TestReconcilerCapsRetryAfter(t *testing.T) {
Timeout: time.Second,
MaxAttempts: 2,
MaxInFlight: 1,
MaxSize: 100,
Retry: RetryConfig{Initial: 50 * time.Millisecond, Max: 2 * time.Second},
}, ports, Runtime{Clock: clock, Sleeper: sleeper})
if err != nil {
@ -371,6 +378,7 @@ func TestReconcilerAppliesAttemptTimeoutToEveryPort(t *testing.T) {
Timeout: 250 * time.Millisecond,
MaxAttempts: 1,
MaxInFlight: 1,
MaxSize: 100,
}, ports)
if err != nil {
t.Fatalf("NewReconciler(): %v", err)
@ -401,6 +409,7 @@ func TestReconcilerDropsNotificationFanoutWhileFetchIsInFlight(t *testing.T) {
Timeout: time.Second,
MaxAttempts: 1,
MaxInFlight: 1,
MaxSize: 100,
}, ports)
if err != nil {
t.Fatalf("NewReconciler(): %v", err)
@ -456,6 +465,7 @@ func TestReconcilerEnforcesMaxInFlightAcrossRunConsumers(t *testing.T) {
Timeout: time.Second,
MaxAttempts: 1,
MaxInFlight: 1,
MaxSize: 100,
}, ports)
if err != nil {
t.Fatalf("NewReconciler(): %v", err)
@ -501,6 +511,7 @@ func TestReconcilerUsesConfiguredMaxInFlight(t *testing.T) {
Timeout: time.Second,
MaxAttempts: 1,
MaxInFlight: 2,
MaxSize: 100,
}, ports)
if err != nil {
t.Fatalf("NewReconciler(): %v", err)
@ -544,6 +555,7 @@ func TestReconcilerDoesNotRefetchWhenActivitySinkFails(t *testing.T) {
Timeout: time.Second,
MaxAttempts: 3,
MaxInFlight: 1,
MaxSize: 100,
Retry: RetryConfig{Initial: 100 * time.Millisecond, Max: time.Second},
}, ports, Runtime{Clock: clock, Sleeper: sleeper})
if err != nil {
@ -576,6 +588,7 @@ func TestReconcilerDoesNotRetryPermanentAdapterError(t *testing.T) {
Timeout: time.Second,
MaxAttempts: 3,
MaxInFlight: 1,
MaxSize: 100,
Retry: RetryConfig{Initial: time.Second, Max: time.Second},
}, ports, Runtime{Sleeper: sleeper})
if err != nil {
@ -602,7 +615,7 @@ func TestReconcilerDoesNotCallProviderWithoutFetchCapacity(t *testing.T) {
return nil, false, nil
})
reconciler, err := NewReconciler(Config{
UpstreamID: "provider-a", Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1,
UpstreamID: "provider-a", Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1, MaxSize: 100,
}, ports)
if err != nil {
t.Fatalf("NewReconciler(): %v", err)
@ -644,7 +657,7 @@ func TestReconcilerCompletesFetchPermitWithFetchedAndRetainedCounts(t *testing.T
return &recordingFetchPermit{expected: 2, completed: completed}, true, nil
})
reconciler, err := NewReconciler(Config{
UpstreamID: "provider-a", Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1,
UpstreamID: "provider-a", Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1, MaxSize: 100,
}, ports)
if err != nil {
t.Fatalf("NewReconciler(): %v", err)
@ -677,6 +690,7 @@ func TestReconcilerRetriesParserErrorsAsFetchErrors(t *testing.T) {
Timeout: time.Second,
MaxAttempts: 2,
MaxInFlight: 1,
MaxSize: 100,
Retry: RetryConfig{Initial: 100 * time.Millisecond, Max: time.Second},
}, ports, Runtime{Clock: clock, Sleeper: sleeper})
if err != nil {
@ -710,14 +724,15 @@ func TestNewReconcilerRejectsInvalidSchedulingConfiguration(t *testing.T) {
name string
config Config
}{
{name: "missing upstream", config: Config{Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1}},
{name: "negative interval", config: Config{UpstreamID: "a", RequestInterval: -1, Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1}},
{name: "missing timeout", config: Config{UpstreamID: "a", MaxAttempts: 1, MaxInFlight: 1}},
{name: "missing attempts", config: Config{UpstreamID: "a", Timeout: time.Second, MaxInFlight: 1}},
{name: "missing in flight", config: Config{UpstreamID: "a", Timeout: time.Second, MaxAttempts: 1}},
{name: "invalid jitter", config: Config{UpstreamID: "a", Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1, Retry: RetryConfig{Jitter: 101}}},
{name: "initial exceeds max", config: Config{UpstreamID: "a", Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1, Retry: RetryConfig{Initial: time.Second, Max: time.Millisecond}}},
{name: "incomplete retry pair", config: Config{UpstreamID: "a", Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1, Retry: RetryConfig{Max: time.Second}}},
{name: "missing upstream", config: Config{Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1, MaxSize: 100}},
{name: "negative interval", config: Config{UpstreamID: "a", RequestInterval: -1, Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1, MaxSize: 100}},
{name: "missing timeout", config: Config{UpstreamID: "a", MaxAttempts: 1, MaxInFlight: 1, MaxSize: 100}},
{name: "missing attempts", config: Config{UpstreamID: "a", Timeout: time.Second, MaxInFlight: 1, MaxSize: 100}},
{name: "missing in flight", config: Config{UpstreamID: "a", Timeout: time.Second, MaxAttempts: 1, MaxSize: 100}},
{name: "missing max size", config: Config{UpstreamID: "a", Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1}},
{name: "invalid jitter", config: Config{UpstreamID: "a", Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1, MaxSize: 100, Retry: RetryConfig{Jitter: 101}}},
{name: "initial exceeds max", config: Config{UpstreamID: "a", Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1, MaxSize: 100, Retry: RetryConfig{Initial: time.Second, Max: time.Millisecond}}},
{name: "incomplete retry pair", config: Config{UpstreamID: "a", Timeout: time.Second, MaxAttempts: 1, MaxInFlight: 1, MaxSize: 100, Retry: RetryConfig{Max: time.Second}}},
}
for _, tt := range tests {

170
internal/domain/activitypool/activity_contracts_test.go Normal file
View File

@ -0,0 +1,170 @@
package activitypool
import (
"context"
"errors"
"testing"
"time"
proxyDomain "proxy-pool/internal/domain/proxy"
)
func TestMemoryPoolAppliesHealthTransitionsAndRejectsStaleObservation(t *testing.T) {
t.Parallel()
now := time.Date(2026, 7, 29, 15, 0, 0, 0, time.UTC)
pool := NewMemoryPool()
inserted, err := pool.UpsertFetched(context.Background(), "provider-a", FetchedBatch{
ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 10,
Proxies: []proxyDomain.Proxy{{
ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080,
State: proxyDomain.StateFetched,
}},
})
if err != nil || inserted.Inserted != 1 {
t.Fatalf("UpsertFetched() = %+v, %v", inserted, err)
}
checking, err := pool.ApplyHealth(context.Background(), HealthUpdate{
ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateChecking,
})
if err != nil || checking.State != proxyDomain.StateChecking {
t.Fatalf("ApplyHealth(checking) = %+v, %v", checking, err)
}
available, err := pool.ApplyHealth(context.Background(), HealthUpdate{
ProxyID: "proxy-a", CheckedAt: now.Add(2 * time.Second),
NextState: proxyDomain.StateAvailable, Latency: 25 * time.Millisecond,
})
if err != nil || available.State != proxyDomain.StateAvailable ||
available.Proxy.LastCheckedAt == nil || !available.Proxy.LastCheckedAt.Equal(now.Add(2*time.Second)) ||
available.Proxy.LastSuccessAt == nil || !available.Proxy.LastSuccessAt.Equal(now.Add(2*time.Second)) ||
available.Proxy.Latency != 25*time.Millisecond {
t.Fatalf("ApplyHealth(available) = %+v, %v", available, err)
}
_, err = pool.ApplyHealth(context.Background(), HealthUpdate{
ProxyID: "proxy-a", CheckedAt: now.Add(time.Second), NextState: proxyDomain.StateSuspect,
})
if !errors.Is(err, ErrStaleHealthUpdate) {
t.Fatalf("ApplyHealth(stale) error = %v, want ErrStaleHealthUpdate", err)
}
replayed, err := pool.ApplyHealth(context.Background(), HealthUpdate{
ProxyID: "proxy-a", CheckedAt: now.Add(2 * time.Second),
NextState: proxyDomain.StateAvailable, Latency: time.Second,
})
if err != nil || replayed.Proxy.Latency != 25*time.Millisecond {
t.Fatalf("ApplyHealth(idempotent replay) = %+v, %v", replayed, err)
}
_, err = pool.ApplyHealth(context.Background(), HealthUpdate{
ProxyID: "proxy-a", CheckedAt: now.Add(2 * time.Second), NextState: proxyDomain.StateSuspect,
})
if !errors.Is(err, ErrStaleHealthUpdate) {
t.Fatalf("ApplyHealth(conflicting replay) error = %v, want ErrStaleHealthUpdate", err)
}
}
func TestMemoryPoolRejectsInvalidHealthUpdates(t *testing.T) {
t.Parallel()
now := time.Date(2026, 7, 29, 15, 0, 0, 0, time.UTC)
pool := NewMemoryPool()
if _, err := pool.UpsertFetched(context.Background(), "provider-a", FetchedBatch{
ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 10,
Proxies: []proxyDomain.Proxy{{
ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080,
State: proxyDomain.StateFetched,
}},
}); err != nil {
t.Fatalf("UpsertFetched(): %v", err)
}
tests := []struct {
name string
update HealthUpdate
want error
}{
{name: "missing proxy ID", update: HealthUpdate{CheckedAt: now, NextState: proxyDomain.StateChecking}, want: ErrInvalidHealthUpdate},
{name: "zero observation time", update: HealthUpdate{ProxyID: "proxy-a", NextState: proxyDomain.StateChecking}, want: ErrInvalidHealthUpdate},
{name: "negative latency", update: HealthUpdate{ProxyID: "proxy-a", CheckedAt: now, NextState: proxyDomain.StateChecking, Latency: -1}, want: ErrInvalidHealthUpdate},
{name: "missing entry", update: HealthUpdate{ProxyID: "missing", CheckedAt: now, NextState: proxyDomain.StateChecking}, want: ErrActivityNotFound},
{name: "invalid transition", update: HealthUpdate{ProxyID: "proxy-a", CheckedAt: now, NextState: proxyDomain.StateAvailable}, want: ErrInvalidHealthUpdate},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if _, err := pool.ApplyHealth(context.Background(), tt.update); !errors.Is(err, tt.want) {
t.Fatalf("ApplyHealth() error = %v, want %v", err, tt.want)
}
})
}
}
func TestMemoryPoolRejectsNonPositiveMaxSize(t *testing.T) {
t.Parallel()
_, err := NewMemoryPool().UpsertFetched(context.Background(), "provider-a", FetchedBatch{
ObservedAt: time.Date(2026, 7, 29, 15, 0, 0, 0, time.UTC),
ConfiguredTTL: time.Minute,
})
if !errors.Is(err, ErrInvalidBatch) {
t.Fatalf("UpsertFetched() error = %v, want ErrInvalidBatch", err)
}
}
func TestMemoryPoolEnforcesMaxSizePerIncumbentUpstream(t *testing.T) {
t.Parallel()
now := time.Date(2026, 7, 29, 15, 0, 0, 0, time.UTC)
pool := NewMemoryPool()
result, err := pool.UpsertFetched(context.Background(), "provider-a", FetchedBatch{
ObservedAt: now, ConfiguredTTL: time.Minute, MaxSize: 1,
Proxies: []proxyDomain.Proxy{
{ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080},
{ID: "proxy-b", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.11", Port: 8080},
},
})
if err != nil {
t.Fatalf("UpsertFetched(): %v", err)
}
if result.Accepted != 2 || result.Inserted != 1 || result.Dropped != 1 {
t.Fatalf("UpsertFetched() = %+v", result)
}
retry, err := pool.UpsertFetched(context.Background(), "provider-a", FetchedBatch{
ObservedAt: now.Add(time.Second), ConfiguredTTL: time.Minute, MaxSize: 2,
Proxies: []proxyDomain.Proxy{
{ID: "proxy-b", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.11", Port: 8080},
},
})
if err != nil || retry.Inserted != 1 || retry.Refreshed != 0 {
t.Fatalf("UpsertFetched(capacity retry) = %+v, %v", retry, err)
}
inventory, err := pool.Inventory(context.Background(), "provider-a", now)
if err != nil || inventory.Managed != 2 {
t.Fatalf("Inventory() = %+v, %v", inventory, err)
}
}
func TestMemoryPoolInventoryAndSweepExpiredAreBounded(t *testing.T) {
t.Parallel()
now := time.Date(2026, 7, 29, 15, 0, 0, 0, time.UTC)
pool := NewMemoryPool()
result, err := pool.UpsertFetched(context.Background(), "provider-a", FetchedBatch{
ObservedAt: now, ConfiguredTTL: time.Second, MaxSize: 2,
Proxies: []proxyDomain.Proxy{
{ID: "proxy-a", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080},
{ID: "proxy-b", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.11", Port: 8080},
},
})
if err != nil || result.Inserted != 2 {
t.Fatalf("UpsertFetched() = %+v, %v", result, err)
}
inventory, err := pool.Inventory(context.Background(), "provider-a", now.Add(2*time.Second))
if err != nil || inventory.Managed != 0 {
t.Fatalf("Inventory(expired) = %+v, %v", inventory, err)
}
first, err := pool.SweepExpired(context.Background(), now.Add(2*time.Second), 1)
if err != nil || first != 1 {
t.Fatalf("SweepExpired(first) = %d, %v", first, err)
}
second, err := pool.SweepExpired(context.Background(), now.Add(2*time.Second), 1)
if err != nil || second != 1 {
t.Fatalf("SweepExpired(second) = %d, %v", second, err)
}
}

190
internal/domain/activitypool/pool.go
View File

@ -16,7 +16,14 @@ import (
const defaultIdempotencyTTL = 5 * time.Minute
var ErrInvalidBatch = errors.New("invalid activity pool batch")
var (
ErrInvalidBatch = errors.New("invalid activity pool batch")
ErrInvalidHealthUpdate = errors.New("invalid activity pool health update")
ErrActivityNotFound = errors.New("activity pool proxy not found")
ErrStaleHealthUpdate = errors.New("stale activity pool health update")
ErrInvalidInventory = errors.New("invalid activity pool inventory query")
ErrInvalidMaintenance = errors.New("invalid activity pool maintenance request")
)
// FetchedBatch describes one ephemeral provider response. Proxies without a
// usable expiry are dropped because this pool is intentionally rebuildable.
@ -24,6 +31,7 @@ type FetchedBatch struct {
ObservedAt time.Time
ConfiguredTTL time.Duration
AllocationSafetyMargin time.Duration
MaxSize int
Proxies []proxyDomain.Proxy
}
@ -40,6 +48,30 @@ type Upserter interface {
UpsertFetched(context.Context, string, FetchedBatch) (UpsertResult, error)
}
type HealthUpdate struct {
ProxyID string
CheckedAt time.Time
NextState proxyDomain.State
Latency time.Duration
}
type Inventory struct {
UpstreamID string
Managed int
}
type HealthStore interface {
ApplyHealth(context.Context, HealthUpdate) (Entry, error)
}
type InventoryReader interface {
Inventory(context.Context, string, time.Time) (Inventory, error)
}
type Maintainer interface {
SweepExpired(context.Context, time.Time, int) (int, error)
}
type Entry struct {
Proxy proxyDomain.Proxy
UsableUntil time.Time
@ -65,6 +97,9 @@ type idempotencyEntry struct {
var (
_ Upserter = (*MemoryPool)(nil)
_ HealthStore = (*MemoryPool)(nil)
_ InventoryReader = (*MemoryPool)(nil)
_ Maintainer = (*MemoryPool)(nil)
_ extractionDomain.Store = (*MemoryPool)(nil)
_ ownershipDomain.Repository = (*MemoryPool)(nil)
)
@ -83,7 +118,7 @@ func (p *MemoryPool) UpsertFetched(ctx context.Context, upstreamID string, batch
if err := ctx.Err(); err != nil {
return result, err
}
if p == nil || upstreamID == "" || batch.ObservedAt.IsZero() || batch.ConfiguredTTL < 0 ||
if p == nil || upstreamID == "" || batch.ObservedAt.IsZero() || batch.ConfiguredTTL < 0 || batch.MaxSize <= 0 ||
batch.AllocationSafetyMargin < 0 ||
(batch.ConfiguredTTL > 0 && batch.AllocationSafetyMargin >= batch.ConfiguredTTL) {
return result, ErrInvalidBatch
@ -100,6 +135,12 @@ func (p *MemoryPool) UpsertFetched(ctx context.Context, upstreamID string, batch
return result, err
}
p.purgeExpiredLocked(batch.ObservedAt)
managedByUpstream := make(map[string]int)
for _, entry := range p.entries {
if managedActivityState(entry.State) {
managedByUpstream[entry.Proxy.SourceUpstream]++
}
}
seenIDs := make(map[string]string, len(batch.Proxies))
for _, candidate := range batch.Proxies {
if !validProxyIdentity(candidate) {
@ -165,6 +206,10 @@ func (p *MemoryPool) UpsertFetched(ctx context.Context, upstreamID string, batch
}
continue
}
if managedByUpstream[upstreamID] >= batch.MaxSize {
result.Dropped++
continue
}
if candidate.ID == "" {
candidate.ID = stableProxyID(key)
@ -175,11 +220,119 @@ func (p *MemoryPool) UpsertFetched(ctx context.Context, upstreamID string, batch
State: candidate.State,
}
p.keyByID[candidate.ID] = key
if managedActivityState(candidate.State) {
managedByUpstream[upstreamID]++
}
result.Inserted++
}
return result, nil
}
func (p *MemoryPool) ApplyHealth(ctx context.Context, update HealthUpdate) (Entry, error) {
if ctx == nil {
return Entry{}, ErrInvalidHealthUpdate
}
if err := ctx.Err(); err != nil {
return Entry{}, err
}
if p == nil || update.ProxyID == "" || update.CheckedAt.IsZero() || update.NextState == "" || update.Latency < 0 {
return Entry{}, ErrInvalidHealthUpdate
}
p.mu.Lock()
defer p.mu.Unlock()
if err := ctx.Err(); err != nil {
return Entry{}, err
}
p.purgeExpiredLocked(update.CheckedAt)
entry, ok := p.entryByIDLocked(update.ProxyID)
if !ok {
return Entry{}, ErrActivityNotFound
}
if entry.Proxy.LastCheckedAt != nil {
if update.CheckedAt.Before(*entry.Proxy.LastCheckedAt) {
return Entry{}, ErrStaleHealthUpdate
}
if update.CheckedAt.Equal(*entry.Proxy.LastCheckedAt) {
if entry.State != update.NextState {
return Entry{}, ErrStaleHealthUpdate
}
entry.Proxy = cloneProxy(entry.Proxy)
return entry, nil
}
}
if entry.State != update.NextState && !proxyDomain.CanTransition(entry.State, update.NextState) {
return Entry{}, ErrInvalidHealthUpdate
}
checkedAt := update.CheckedAt.UTC()
entry.State = update.NextState
entry.Proxy.State = update.NextState
entry.Proxy.LastCheckedAt = &checkedAt
entry.Proxy.Latency = update.Latency
if update.NextState == proxyDomain.StateAvailable {
lastSuccessAt := checkedAt
entry.Proxy.LastSuccessAt = &lastSuccessAt
}
p.setEntryByIDLocked(update.ProxyID, entry)
entry.Proxy = cloneProxy(entry.Proxy)
return entry, nil
}
func (p *MemoryPool) Inventory(ctx context.Context, upstreamID string, now time.Time) (Inventory, error) {
result := Inventory{UpstreamID: upstreamID}
if ctx == nil {
return result, ErrInvalidInventory
}
if err := ctx.Err(); err != nil {
return result, err
}
if p == nil || upstreamID == "" || now.IsZero() {
return result, ErrInvalidInventory
}
p.mu.Lock()
defer p.mu.Unlock()
if err := ctx.Err(); err != nil {
return result, err
}
for _, entry := range p.entries {
if entry.Proxy.SourceUpstream == upstreamID && managedActivityState(entry.State) &&
entry.Proxy.ExpiresAt != nil && entry.Proxy.ExpiresAt.After(now) {
result.Managed++
}
}
return result, nil
}
func (p *MemoryPool) SweepExpired(ctx context.Context, now time.Time, limit int) (int, error) {
if ctx == nil {
return 0, ErrInvalidMaintenance
}
if err := ctx.Err(); err != nil {
return 0, err
}
if p == nil || now.IsZero() || limit <= 0 {
return 0, ErrInvalidMaintenance
}
p.mu.Lock()
defer p.mu.Unlock()
if err := ctx.Err(); err != nil {
return 0, err
}
expiredIDs := make([]string, 0)
for _, entry := range p.entries {
if entry.Proxy.ExpiresAt != nil && !entry.Proxy.ExpiresAt.After(now) {
expiredIDs = append(expiredIDs, entry.Proxy.ID)
}
}
sort.Strings(expiredIDs)
if len(expiredIDs) > limit {
expiredIDs = expiredIDs[:limit]
}
for _, proxyID := range expiredIDs {
p.removeEntryByIDLocked(proxyID)
}
return len(expiredIDs), nil
}
func (p *MemoryPool) Snapshot(now time.Time) []Entry {
if p == nil {
return nil
@ -473,9 +626,7 @@ func (p *MemoryPool) purgeExpiredLocked(now time.Time) int {
if entry.Proxy.ExpiresAt == nil || entry.Proxy.ExpiresAt.After(now) {
continue
}
delete(p.ownership, entry.Proxy.ID)
delete(p.keyByID, entry.Proxy.ID)
delete(p.entries, key)
p.removeEntryLocked(key, entry)
removed++
}
for key, entry := range p.idempotent {
@ -486,6 +637,25 @@ func (p *MemoryPool) purgeExpiredLocked(now time.Time) int {
return removed
}
func (p *MemoryPool) removeEntryByIDLocked(proxyID string) {
key, ok := p.keyByID[proxyID]
if !ok {
return
}
entry, ok := p.entries[key]
if !ok {
delete(p.keyByID, proxyID)
return
}
p.removeEntryLocked(key, entry)
}
func (p *MemoryPool) removeEntryLocked(key string, entry Entry) {
delete(p.ownership, entry.Proxy.ID)
delete(p.keyByID, entry.Proxy.ID)
delete(p.entries, key)
}
func (p *MemoryPool) entryByIDLocked(proxyID string) (Entry, bool) {
key, ok := p.keyByID[proxyID]
if !ok {
@ -554,6 +724,16 @@ func validProxyIdentity(candidate proxyDomain.Proxy) bool {
}
}
func managedActivityState(state proxyDomain.State) bool {
switch state {
case proxyDomain.StateFetched, proxyDomain.StateChecking, proxyDomain.StateAvailable,
proxyDomain.StateSuspect, proxyDomain.StateDraining:
return true
default:
return false
}
}
func cloneProxy(candidate proxyDomain.Proxy) proxyDomain.Proxy {
if candidate.ExpiresAt != nil {
value := *candidate.ExpiresAt

16
internal/domain/activitypool/pool_test.go
View File

@ -19,6 +19,7 @@ func TestMemoryPoolUpsertAppliesProviderTTLAndRefreshesWithoutGrowth(t *testing.
ObservedAt: now,
ConfiguredTTL: 30 * time.Second,
AllocationSafetyMargin: 3 * time.Second,
MaxSize: 100,
Proxies: []proxyDomain.Proxy{{
Scheme: proxyDomain.SchemeHTTP,
Host: "192.0.2.10",
@ -70,7 +71,7 @@ func TestMemoryPoolCrossProviderDuplicateDoesNotReplaceSourceLifecycle(t *testin
State: proxyDomain.StateAvailable,
}
first, err := pool.UpsertFetched(context.Background(), "provider-a", FetchedBatch{
ObservedAt: now, ConfiguredTTL: 30 * time.Second,
ObservedAt: now, ConfiguredTTL: 30 * time.Second, MaxSize: 100,
AllocationSafetyMargin: 3 * time.Second, Proxies: []proxyDomain.Proxy{proxy},
})
if err != nil || first.Inserted != 1 {
@ -78,7 +79,7 @@ func TestMemoryPoolCrossProviderDuplicateDoesNotReplaceSourceLifecycle(t *testin
}
duplicate, err := pool.UpsertFetched(context.Background(), "provider-b", FetchedBatch{
ObservedAt: now.Add(time.Second), ConfiguredTTL: 5 * time.Minute,
ObservedAt: now.Add(time.Second), ConfiguredTTL: 5 * time.Minute, MaxSize: 100,
AllocationSafetyMargin: 10 * time.Second, Proxies: []proxyDomain.Proxy{proxy},
})
if err != nil || duplicate.Inserted != 0 || duplicate.Refreshed != 1 {
@ -99,7 +100,7 @@ func TestMemoryPoolCrossProviderDuplicateDoesNotReplaceSourceLifecycle(t *testin
}
afterExpiry, err := pool.UpsertFetched(context.Background(), "provider-b", FetchedBatch{
ObservedAt: now.Add(31 * time.Second), ConfiguredTTL: 5 * time.Minute,
ObservedAt: now.Add(31 * time.Second), ConfiguredTTL: 5 * time.Minute, MaxSize: 100,
AllocationSafetyMargin: 10 * time.Second, Proxies: []proxyDomain.Proxy{proxy},
})
if err != nil || afterExpiry.Inserted != 1 {
@ -116,7 +117,7 @@ func TestMemoryPoolRefreshPreservesRuntimeStateAndHealth(t *testing.T) {
checkedAt := now.Add(-time.Second)
pool := NewMemoryPool()
first, err := pool.UpsertFetched(context.Background(), "provider-a", FetchedBatch{
ObservedAt: now, ConfiguredTTL: 30 * time.Second,
ObservedAt: now, ConfiguredTTL: 30 * time.Second, MaxSize: 100,
Proxies: []proxyDomain.Proxy{{
Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080,
State: proxyDomain.StateAvailable, LastCheckedAt: &checkedAt,
@ -128,7 +129,7 @@ func TestMemoryPoolRefreshPreservesRuntimeStateAndHealth(t *testing.T) {
initial := pool.Snapshot(now)[0]
second, err := pool.UpsertFetched(context.Background(), "provider-a", FetchedBatch{
ObservedAt: now.Add(10 * time.Second), ConfiguredTTL: 30 * time.Second,
ObservedAt: now.Add(10 * time.Second), ConfiguredTTL: 30 * time.Second, MaxSize: 100,
Proxies: []proxyDomain.Proxy{{
Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8080,
State: proxyDomain.StateFetched,
@ -154,6 +155,7 @@ func TestMemoryPoolDropsCandidatesWithoutUsableTTL(t *testing.T) {
result, err := pool.UpsertFetched(context.Background(), "provider-a", FetchedBatch{
ObservedAt: now,
AllocationSafetyMargin: 3 * time.Second,
MaxSize: 100,
Proxies: []proxyDomain.Proxy{
{Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8001},
{Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.11", Port: 8002, ExpiresAt: &expired},
@ -174,6 +176,7 @@ func TestMemoryPoolRejectsInvalidBatchAtomically(t *testing.T) {
result, err := pool.UpsertFetched(context.Background(), "provider-a", FetchedBatch{
ObservedAt: now,
ConfiguredTTL: time.Minute,
MaxSize: 100,
Proxies: []proxyDomain.Proxy{
{Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.10", Port: 8001},
{Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.11", Port: 8002, SourceUpstream: "provider-b"},
@ -397,7 +400,7 @@ func TestMemoryPoolExpireUsesStableProxyIDOrderAndLimit(t *testing.T) {
{ID: "proxy-b", Scheme: proxyDomain.SchemeHTTP, Host: "192.0.2.2", Port: 8002, State: proxyDomain.StateAvailable},
}
if _, err := pool.UpsertFetched(context.Background(), "provider-a", FetchedBatch{
ObservedAt: now, ConfiguredTTL: 10 * time.Minute, Proxies: proxies,
ObservedAt: now, ConfiguredTTL: 10 * time.Minute, MaxSize: 100, Proxies: proxies,
}); err != nil {
t.Fatalf("UpsertFetched(): %v", err)
}
@ -439,6 +442,7 @@ func poolWithOneProxy(t *testing.T, now time.Time) *MemoryPool {
ObservedAt: now,
ConfiguredTTL: 30 * time.Second,
AllocationSafetyMargin: 3 * time.Second,
MaxSize: 100,
Proxies: []proxyDomain.Proxy{{
Scheme: proxyDomain.SchemeHTTP,
Host: "192.0.2.10",