diff --git a/control-plane/cmd/controlplane/main.go b/control-plane/cmd/controlplane/main.go index 72ce852..d2fa7f6 100644 --- a/control-plane/cmd/controlplane/main.go +++ b/control-plane/cmd/controlplane/main.go @@ -22,6 +22,7 @@ import ( "github.com/openinfra/network/internal/dashboard" "github.com/openinfra/network/internal/orchestrator" "github.com/openinfra/network/internal/providerjoin" + "github.com/openinfra/network/internal/resourcemarket" "github.com/openinfra/network/internal/scheduler" "github.com/openinfra/network/internal/wireguard" "github.com/openinfra/network/internal/workloadapi" @@ -134,6 +135,10 @@ func run() error { if err != nil { return fmt.Errorf("configure Provider Agent client: %w", err) } + // Keeps pallet-resource-market's on-chain Offers in sync with each + // schedulable provider's declared total capacity (issue #15). + marketReconciler := resourcemarket.NewReconciler(directory, marketBridge{registrar: registrar, chain: chainClient}, resourcemarket.DefaultReconcilerConfig()) + go marketReconciler.Run(ctx) ranker := scheduler.NewRanker(scheduler.DefaultMaxReputationScore, scheduler.DefaultDefaultReputationScore) worker := orchestrator.NewWorker(workloadRepository, directory, registrar, agentClient, ranker) worker.SetReputationSource(chainClient) @@ -247,6 +252,29 @@ func serverOptions(address string) ([]grpc.ServerOption, error) { return []grpc.ServerOption{grpc.Creds(credentials.NewTLS(configuration))}, nil } +// marketBridge combines *blockchainbridge.Registrar's write methods with +// *blockchainbridge.RPCClient's read methods into the single +// resourcemarket.Market surface -- they are genuinely two different +// receiver types in blockchainbridge (signing/submitting vs. querying +// storage), not an arbitrary split introduced here. +type marketBridge struct { + registrar *blockchainbridge.Registrar + chain *blockchainbridge.RPCClient +} + +func (b marketBridge) AnnounceOfferFor(ctx context.Context, provider [32]byte, offer blockchainbridge.ResourceOffer) error { + return b.registrar.AnnounceOfferFor(ctx, provider, offer) +} +func (b marketBridge) RemoveOfferFor(ctx context.Context, provider [32]byte) error { + return b.registrar.RemoveOfferFor(ctx, provider) +} +func (b marketBridge) FinalizedOffer(ctx context.Context, provider [32]byte, blockHash string) (blockchainbridge.ResourceOffer, bool, error) { + return b.chain.FinalizedOffer(ctx, provider, blockHash) +} +func (b marketBridge) FinalizedHead(ctx context.Context) (string, error) { + return b.chain.FinalizedHead(ctx) +} + func envOrDefault(name, fallback string) string { if value := os.Getenv(name); value != "" { return value diff --git a/control-plane/internal/resourcemarket/reconciler.go b/control-plane/internal/resourcemarket/reconciler.go new file mode 100644 index 0000000..bfbefbc --- /dev/null +++ b/control-plane/internal/resourcemarket/reconciler.go @@ -0,0 +1,163 @@ +// Package resourcemarket keeps pallet-resource-market's on-chain Offers in +// sync with each provider's actual capacity, closing the gap issue #15 +// found: the pallet existed with zero Go-side integration, so an offer's +// on-chain state could drift arbitrarily far from reality. +package resourcemarket + +import ( + "context" + "crypto/ed25519" + "log/slog" + "time" + + "github.com/openinfra/network/internal/agentmanager" + "github.com/openinfra/network/internal/blockchainbridge" + "github.com/openinfra/network/internal/workloadapi" +) + +// Directory is the same live provider view the scheduler ranks against. +type Directory interface { + ListSchedulableProviders(ctx context.Context) ([]agentmanager.SchedulableProvider, error) +} + +// Market is the on-chain publish/read surface, satisfied by +// *blockchainbridge.Registrar (writes) and *blockchainbridge.RPCClient +// (reads) together -- main.go wires the same chainClient/registrar pair +// already used everywhere else. +type Market interface { + AnnounceOfferFor(ctx context.Context, provider [32]byte, offer blockchainbridge.ResourceOffer) error + RemoveOfferFor(ctx context.Context, provider [32]byte) error + FinalizedOffer(ctx context.Context, provider [32]byte, blockHash string) (blockchainbridge.ResourceOffer, bool, error) + FinalizedHead(ctx context.Context) (string, error) +} + +type ReconcilerConfig struct { + Interval time.Duration +} + +func DefaultReconcilerConfig() ReconcilerConfig { + return ReconcilerConfig{Interval: 30 * time.Second} +} + +// Reconciler publishes/updates each currently-schedulable provider's offer +// from its declared *total* capacity (the ceiling the scheduler's atomic +// capacity check already uses -- not the fast-changing "available" figure, +// which stays off-chain in Redis by design; publishing that to the chain +// on every workload placement would be both far too chatty for a +// blockchain and pointless, since AGENTS.md already treats Redis as the +// authoritative reconstructible store for exactly this kind of data), and +// withdraws offers for providers that drop out of that set. +// +// Withdrawal tracking (offering, below) is in-memory, not persisted: after +// a restart, a provider that vanished during the outage keeps a stale +// offer for up to one reconcile interval before it's noticed and removed. +// This is a deliberately bounded, self-healing gap, not a security +// control -- no scheduling decision consults on-chain offers yet (that +// integration is separate, still-open work for #15), so a briefly-stale +// offer has no live consequence today. +type Reconciler struct { + directory Directory + market Market + cfg ReconcilerConfig + // offering maps ProviderID to the 32-byte key used to publish its + // offer, for every provider this process believes currently holds + // one -- needed for withdrawal, since a provider that drops out of + // ListSchedulableProviders can no longer be looked up there. + offering map[string][32]byte +} + +func NewReconciler(directory Directory, market Market, cfg ReconcilerConfig) *Reconciler { + if cfg.Interval <= 0 { + cfg.Interval = DefaultReconcilerConfig().Interval + } + return &Reconciler{directory: directory, market: market, cfg: cfg, offering: make(map[string][32]byte)} +} + +func (r *Reconciler) Run(ctx context.Context) { + ticker := time.NewTicker(r.cfg.Interval) + defer ticker.Stop() + for { + r.ReconcileOnce(ctx) + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +// ReconcileOnce processes a single pass and returns without waiting on the +// ticker -- exported so tests can drive it deterministically. +func (r *Reconciler) ReconcileOnce(ctx context.Context) { + providers, err := r.directory.ListSchedulableProviders(ctx) + if err != nil { + slog.Error("resourcemarket: failed to list schedulable providers", "error", err) + return + } + head, err := r.market.FinalizedHead(ctx) + if err != nil { + slog.Error("resourcemarket: failed to resolve finalized head", "error", err) + return + } + + seen := make(map[string]struct{}, len(providers)) + for _, provider := range providers { + if len(provider.PublicKey) != ed25519.PublicKeySize || provider.Capabilities == nil { + continue + } + var key [32]byte + copy(key[:], provider.PublicKey) + seen[provider.ProviderID] = struct{}{} + + desired := blockchainbridge.ResourceOffer{ + CPUMillicores: clampToUint32(workloadapi.CPUCoresToMillicores(provider.Capabilities.CpuTotal)), + RAMMB: uint64(provider.Capabilities.RamTotalMb), + StorageGB: uint64(provider.Capabilities.StorageTotalGb), + } + current, found, err := r.market.FinalizedOffer(ctx, key, head) + if err != nil { + slog.Warn("resourcemarket: finalized offer read failed; will retry next pass", "provider_id", provider.ProviderID, "error", err) + continue + } + if found && offersEqual(current, desired) { + r.offering[provider.ProviderID] = key + continue + } + if err := r.market.AnnounceOfferFor(ctx, key, desired); err != nil { + slog.Error("resourcemarket: announce_offer_for failed", "provider_id", provider.ProviderID, "error", err) + continue + } + r.offering[provider.ProviderID] = key + } + + for providerID, key := range r.offering { + if _, stillSchedulable := seen[providerID]; stillSchedulable { + continue + } + if err := r.market.RemoveOfferFor(ctx, key); err != nil { + slog.Error("resourcemarket: remove_offer_for failed; will retry next pass", "provider_id", providerID, "error", err) + continue + } + delete(r.offering, providerID) + } +} + +// offersEqual ignores Capabilities: this reconciler never sets it (nothing +// downstream reads it yet -- capability tags are future work), so +// comparing it would force a redundant announce_offer_for on every pass. +func offersEqual(a, b blockchainbridge.ResourceOffer) bool { + return a.CPUMillicores == b.CPUMillicores && a.RAMMB == b.RAMMB && a.StorageGB == b.StorageGB +} + +// clampToUint32 protects the on-chain u32 field from a CPU total large +// enough to overflow it (millicores of ~4.29 million cores) rather than +// silently wrapping. +func clampToUint32(value int64) uint32 { + if value < 0 { + return 0 + } + if value > int64(^uint32(0)) { + return ^uint32(0) + } + return uint32(value) +} diff --git a/control-plane/internal/resourcemarket/reconciler_test.go b/control-plane/internal/resourcemarket/reconciler_test.go new file mode 100644 index 0000000..f3cbcd1 --- /dev/null +++ b/control-plane/internal/resourcemarket/reconciler_test.go @@ -0,0 +1,211 @@ +package resourcemarket + +import ( + "context" + "errors" + "testing" + + "github.com/openinfra/network/internal/agentmanager" + "github.com/openinfra/network/internal/blockchainbridge" + sharedv1 "github.com/openinfra/network/protocol/generated/go/shared/v1" +) + +type fakeDirectory struct { + providers []agentmanager.SchedulableProvider + err error +} + +func (d fakeDirectory) ListSchedulableProviders(context.Context) ([]agentmanager.SchedulableProvider, error) { + return d.providers, d.err +} + +type fakeMarket struct { + head string + headErr error + offers map[[32]byte]blockchainbridge.ResourceOffer + readErr map[[32]byte]error + announceCalls []blockchainbridge.ResourceOffer + announceErr error + removeCalls [][32]byte + removeErr error +} + +func newFakeMarket() *fakeMarket { + return &fakeMarket{head: "0xhead", offers: make(map[[32]byte]blockchainbridge.ResourceOffer), readErr: make(map[[32]byte]error)} +} + +func (m *fakeMarket) FinalizedHead(context.Context) (string, error) { return m.head, m.headErr } + +func (m *fakeMarket) FinalizedOffer(_ context.Context, provider [32]byte, _ string) (blockchainbridge.ResourceOffer, bool, error) { + if err, ok := m.readErr[provider]; ok { + return blockchainbridge.ResourceOffer{}, false, err + } + offer, found := m.offers[provider] + return offer, found, nil +} + +func (m *fakeMarket) AnnounceOfferFor(_ context.Context, provider [32]byte, offer blockchainbridge.ResourceOffer) error { + if m.announceErr != nil { + return m.announceErr + } + m.announceCalls = append(m.announceCalls, offer) + m.offers[provider] = offer + return nil +} + +func (m *fakeMarket) RemoveOfferFor(_ context.Context, provider [32]byte) error { + if m.removeErr != nil { + return m.removeErr + } + m.removeCalls = append(m.removeCalls, provider) + delete(m.offers, provider) + return nil +} + +func testProvider(id string, seed byte, cpuTotal float32, ramTotalMb, storageTotalGb int64) agentmanager.SchedulableProvider { + key := make([]byte, 32) + for i := range key { + key[i] = seed + } + return agentmanager.SchedulableProvider{ + RegisteredProvider: agentmanager.RegisteredProvider{ProviderID: id, PublicKey: key, AgentEndpoint: "https://" + id}, + Capabilities: &sharedv1.ResourceCapability{ + CpuTotal: cpuTotal, CpuAvailable: cpuTotal, + RamTotalMb: ramTotalMb, RamAvailableMb: ramTotalMb, + StorageTotalGb: storageTotalGb, StorageAvailableGb: storageTotalGb, + }, + } +} + +func TestReconcileOnceAnnouncesANewProvider(t *testing.T) { + directory := fakeDirectory{providers: []agentmanager.SchedulableProvider{testProvider("p1", 1, 2, 4096, 100)}} + market := newFakeMarket() + reconciler := NewReconciler(directory, market, ReconcilerConfig{}) + + reconciler.ReconcileOnce(context.Background()) + + if len(market.announceCalls) != 1 { + t.Fatalf("expected exactly one announce call, got %d", len(market.announceCalls)) + } + got := market.announceCalls[0] + if got.CPUMillicores != 2000 || got.RAMMB != 4096 || got.StorageGB != 100 { + t.Fatalf("unexpected offer: %+v", got) + } +} + +func TestReconcileOnceSkipsAnAlreadyCorrectOffer(t *testing.T) { + provider := testProvider("p1", 1, 2, 4096, 100) + var key [32]byte + copy(key[:], provider.PublicKey) + directory := fakeDirectory{providers: []agentmanager.SchedulableProvider{provider}} + market := newFakeMarket() + market.offers[key] = blockchainbridge.ResourceOffer{CPUMillicores: 2000, RAMMB: 4096, StorageGB: 100} + reconciler := NewReconciler(directory, market, ReconcilerConfig{}) + + reconciler.ReconcileOnce(context.Background()) + + if len(market.announceCalls) != 0 { + t.Fatalf("expected no announce call for an already-correct offer, got %d", len(market.announceCalls)) + } +} + +func TestReconcileOnceUpdatesAChangedOffer(t *testing.T) { + provider := testProvider("p1", 1, 4, 8192, 200) // capacity grew + var key [32]byte + copy(key[:], provider.PublicKey) + directory := fakeDirectory{providers: []agentmanager.SchedulableProvider{provider}} + market := newFakeMarket() + market.offers[key] = blockchainbridge.ResourceOffer{CPUMillicores: 2000, RAMMB: 4096, StorageGB: 100} // stale, smaller + reconciler := NewReconciler(directory, market, ReconcilerConfig{}) + + reconciler.ReconcileOnce(context.Background()) + + if len(market.announceCalls) != 1 || market.announceCalls[0].CPUMillicores != 4000 { + t.Fatalf("expected the offer to be updated to the new capacity, calls=%+v", market.announceCalls) + } +} + +func TestReconcileWithdrawsAProviderThatDropsOutOfTheSchedulableSet(t *testing.T) { + provider := testProvider("p1", 1, 2, 4096, 100) + var key [32]byte + copy(key[:], provider.PublicKey) + market := newFakeMarket() + reconciler := NewReconciler(fakeDirectory{providers: []agentmanager.SchedulableProvider{provider}}, market, ReconcilerConfig{}) + + reconciler.ReconcileOnce(context.Background()) + if len(market.announceCalls) != 1 { + t.Fatalf("expected the first pass to announce, got %d calls", len(market.announceCalls)) + } + + // Second pass: the provider is gone (deregistered, or heartbeat went stale). + reconciler.directory = fakeDirectory{providers: nil} + reconciler.ReconcileOnce(context.Background()) + + if len(market.removeCalls) != 1 || market.removeCalls[0] != key { + t.Fatalf("expected the vanished provider's offer to be withdrawn, removeCalls=%+v", market.removeCalls) + } + if _, stillTracked := reconciler.offering[provider.ProviderID]; stillTracked { + t.Fatal("expected the withdrawn provider to be forgotten") + } +} + +func TestReconcileOnceSkipsProvidersWithoutAUsableKeyOrCapabilities(t *testing.T) { + noKey := agentmanager.SchedulableProvider{RegisteredProvider: agentmanager.RegisteredProvider{ProviderID: "no-key", PublicKey: []byte{1, 2, 3}}, Capabilities: &sharedv1.ResourceCapability{CpuTotal: 1}} + noCapabilities := agentmanager.SchedulableProvider{RegisteredProvider: agentmanager.RegisteredProvider{ProviderID: "no-caps", PublicKey: make([]byte, 32)}} + directory := fakeDirectory{providers: []agentmanager.SchedulableProvider{noKey, noCapabilities}} + market := newFakeMarket() + reconciler := NewReconciler(directory, market, ReconcilerConfig{}) + + reconciler.ReconcileOnce(context.Background()) + + if len(market.announceCalls) != 0 { + t.Fatalf("expected no announce calls for unusable candidates, got %d", len(market.announceCalls)) + } +} + +func TestReconcileOnceToleratesADirectoryFailureWithoutPanicking(t *testing.T) { + directory := fakeDirectory{err: errors.New("redis unavailable")} + market := newFakeMarket() + reconciler := NewReconciler(directory, market, ReconcilerConfig{}) + reconciler.ReconcileOnce(context.Background()) // must not panic + if len(market.announceCalls) != 0 { + t.Fatal("expected no announce calls when the directory read failed") + } +} + +func TestReconcileOnceKeepsRetryingAFailedAnnounce(t *testing.T) { + provider := testProvider("p1", 1, 2, 4096, 100) + directory := fakeDirectory{providers: []agentmanager.SchedulableProvider{provider}} + market := newFakeMarket() + market.announceErr = errors.New("chain unavailable") + reconciler := NewReconciler(directory, market, ReconcilerConfig{}) + + reconciler.ReconcileOnce(context.Background()) + if _, tracked := reconciler.offering[provider.ProviderID]; tracked { + t.Fatal("a failed announce must not be recorded as offering, or a real removal would never be attempted") + } + + market.announceErr = nil + reconciler.ReconcileOnce(context.Background()) + if len(market.announceCalls) != 1 { + t.Fatalf("expected the retry to succeed on the next pass, got %d calls", len(market.announceCalls)) + } +} + +func TestClampToUint32(t *testing.T) { + cases := []struct { + in int64 + want uint32 + }{ + {-1, 0}, + {0, 0}, + {1000, 1000}, + {int64(^uint32(0)), ^uint32(0)}, + {int64(^uint32(0)) + 1, ^uint32(0)}, + } + for _, c := range cases { + if got := clampToUint32(c.in); got != c.want { + t.Errorf("clampToUint32(%d) = %d, want %d", c.in, got, c.want) + } + } +}