From 2156f75f05a3bfa710a2a89bf38c66f5204cf6bb Mon Sep 17 00:00:00 2001 From: FlorianJeandenans Date: Thu, 6 Aug 2026 15:51:39 +0200 Subject: [PATCH] feat(control-plane): reconcile pallet-resource-market offers with live capacity Third and final slice of #15, on top of the delegated calls (#69) and the bridge read/write layer (#70). internal/resourcemarket/reconciler.go: - ReconcileOnce lists the same live schedulable-provider view the scheduler ranks against, computes each provider's desired ResourceOffer from its declared *total* capacity (the ceiling the scheduler's atomic Postgres capacity check already uses -- not the fast-changing "available" figure, which stays off-chain in Redis by design), compares it to the chain's FinalizedOffer, and calls AnnounceOfferFor only when it actually differs. Providers that drop out of the schedulable set (deregistered, or heartbeat went stale) have their offer withdrawn via RemoveOfferFor. - Withdrawal tracking (the offering map) is in-memory, not persisted: after a Control Plane restart, a provider that vanished during the outage keeps a stale on-chain 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, so a briefly-stale offer has no live consequence today. - clampToUint32 protects the pallet's u32 CPU field from a total large enough to overflow it, instead of silently wrapping. Wired into cmd/controlplane/main.go via a marketBridge adapter that combines *blockchainbridge.Registrar's write methods with *blockchainbridge.RPCClient's read methods into the resourcemarket.Market interface the reconciler depends on, run alongside the existing provider-join outbox reconciler. Adds 8 tests: announces a new provider, skips an already-correct offer, updates a changed offer, withdraws a provider that drops out of the schedulable set (and forgets it afterward), skips providers without a usable 32-byte key or capabilities, tolerates a directory read failure without panicking, keeps retrying a failed announce on the next pass, and clampToUint32's boundary cases. Verified (control-plane/): gofmt -l .; go build ./...; go vet ./...; go test ./... -- full suite green against merged main, including this package (0.027s). This closes out #15's three planned slices. Still explicitly out of scope, left for follow-up: the scheduler consulting finalized on-chain offers before selecting a provider, and integer pricing (needs a proto change + consumer analysis). Co-Authored-By: Claude Sonnet 5 --- control-plane/cmd/controlplane/main.go | 28 +++ .../internal/resourcemarket/reconciler.go | 163 ++++++++++++++ .../resourcemarket/reconciler_test.go | 211 ++++++++++++++++++ 3 files changed, 402 insertions(+) create mode 100644 control-plane/internal/resourcemarket/reconciler.go create mode 100644 control-plane/internal/resourcemarket/reconciler_test.go 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) + } + } +}