From 3da836aeff2a99b074a5dc9562a22043633730af Mon Sep 17 00:00:00 2001 From: youfak Date: Wed, 29 Jul 2026 15:03:16 +0800 Subject: [PATCH] feat: add activity health and inventory contracts --- .../controller/extraction/service_test.go | 1 + internal/controller/pool/ownership_test.go | 2 +- internal/controller/provider/reconciler.go | 5 +- .../controller/provider/reconciler_test.go | 37 +++- .../activitypool/activity_contracts_test.go | 170 ++++++++++++++++ internal/domain/activitypool/pool.go | 190 +++++++++++++++++- internal/domain/activitypool/pool_test.go | 16 +- 7 files changed, 397 insertions(+), 24 deletions(-) create mode 100644 internal/domain/activitypool/activity_contracts_test.go diff --git a/internal/controller/extraction/service_test.go b/internal/controller/extraction/service_test.go index 4995473..45c758f 100644 --- a/internal/controller/extraction/service_test.go +++ b/internal/controller/extraction/service_test.go @@ -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, diff --git a/internal/controller/pool/ownership_test.go b/internal/controller/pool/ownership_test.go index ffdb44f..2432695 100644 --- a/internal/controller/pool/ownership_test.go +++ b/internal/controller/pool/ownership_test.go @@ -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) diff --git a/internal/controller/provider/reconciler.go b/internal/controller/provider/reconciler.go index 1514cd2..dcd2358 100644 --- a/internal/controller/provider/reconciler.go +++ b/internal/controller/provider/reconciler.go @@ -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 diff --git a/internal/controller/provider/reconciler_test.go b/internal/controller/provider/reconciler_test.go index b6dc41f..fb29896 100644 --- a/internal/controller/provider/reconciler_test.go +++ b/internal/controller/provider/reconciler_test.go @@ -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 { diff --git a/internal/domain/activitypool/activity_contracts_test.go b/internal/domain/activitypool/activity_contracts_test.go new file mode 100644 index 0000000..44d6272 --- /dev/null +++ b/internal/domain/activitypool/activity_contracts_test.go @@ -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) + } +} diff --git a/internal/domain/activitypool/pool.go b/internal/domain/activitypool/pool.go index a3a4a5d..921f0d5 100644 --- a/internal/domain/activitypool/pool.go +++ b/internal/domain/activitypool/pool.go @@ -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 diff --git a/internal/domain/activitypool/pool_test.go b/internal/domain/activitypool/pool_test.go index c4f785a..0c2e38e 100644 --- a/internal/domain/activitypool/pool_test.go +++ b/internal/domain/activitypool/pool_test.go @@ -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",