From d3576327e9ed3ce1f3d5bf67a4572d117b24fac9 Mon Sep 17 00:00:00 2001 From: FlorianJeandenans Date: Thu, 6 Aug 2026 15:29:11 +0200 Subject: [PATCH] feat(control-plane): read/write pallet-resource-market offers Second slice of #15, on top of the delegated calls from #69. internal/blockchainbridge/resourcemarket.go: - AnnounceOfferFor/RemoveOfferFor submit announce_offer_for/ remove_offer_for, mirroring EnsureActive's proven submission shape exactly (serialize nonce use under the same mutex, sudo-wrap, sign, submit). Not live-verified: the locally running dev chain predates this pallet change and I have no way to rebuild/redeploy it in this sandbox (no libclang for openinfra-node), so the encoding is built and unit-tested against the same primitives EnsureActive already uses in production, not against a live extrinsic acceptance. - FinalizedOffer/decodeResourceOffer read pallet-resource-market's Offers map (single-key Blake2_128Concat, same shape as the reputation/validator reads from #46-#49) and decode cpu/ram/storage (fixed u32+u64+u64) plus capabilities (compact-length-prefixed bytes, the same shape decodeAccountIdVec already handles for the validator set). This is live-verified: Offers storage is unchanged by #69, so it ran against the running local dev chain during development -- correct "no offer yet" for every registered provider, no decode errors. ResourceOffer documents this bridge's own unit convention explicitly (CPU in millicores, matching workloadapi.CPUCoresToMillicores and the scheduler; RAM in MB, storage in GB, matching ResourceCapability on the wire) since the pallet's u32/u64 fields carry no unit themselves -- issue #15 asks for units to be defined, not left implicit. Adds 5 tests: fixed-field + capabilities round trip across several shapes including empty capabilities; truncated and trailing-byte inputs both rejected (a decoder that silently ignores trailing bytes would also silently accept a corrupted encoding); encodeBoundedBytes matches the existing compact decoder; storage key is deterministic and provider-distinguishing. Verified (control-plane/): go build ./...; go vet ./...; gofmt -l .; go test ./... (full suite green). Still open in #15: a reconciler that actually calls Announce/ RemoveOfferFor as provider capacity/status changes, the scheduler checking finalized on-chain offers before selecting a provider, and integer pricing (needs a proto change + consumer analysis, out of scope for a bridge-layer change). Co-Authored-By: Claude Sonnet 5 --- .../blockchainbridge/resourcemarket.go | 165 ++++++++++++++++++ .../blockchainbridge/resourcemarket_test.go | 80 +++++++++ 2 files changed, 245 insertions(+) create mode 100644 control-plane/internal/blockchainbridge/resourcemarket.go create mode 100644 control-plane/internal/blockchainbridge/resourcemarket_test.go diff --git a/control-plane/internal/blockchainbridge/resourcemarket.go b/control-plane/internal/blockchainbridge/resourcemarket.go new file mode 100644 index 0000000..eb70f25 --- /dev/null +++ b/control-plane/internal/blockchainbridge/resourcemarket.go @@ -0,0 +1,165 @@ +package blockchainbridge + +import ( + "context" + "encoding/binary" + "errors" +) + +const ( + resourceMarketPalletIndex = 11 + announceOfferForCallIndex = 2 + removeOfferForCallIndex = 3 +) + +// ResourceOffer mirrors pallet-resource-market's on-chain ResourceOffer. +// Units are this bridge's own documented convention -- the pallet's u32/ +// u64 fields carry no unit by themselves, and issue #15 explicitly asks +// for units to be defined, not left implicit: +// +// - CPUMillicores: millicores (1000 = one whole core), matching the same +// conversion workloadapi.CPUCoresToMillicores and the scheduler use +// for every other CPU comparison in this codebase. +// - RAMMB: megabytes, matching ResourceCapability.ram_*_mb on the wire. +// - StorageGB: gigabytes, matching ResourceCapability.storage_*_gb. +type ResourceOffer struct { + CPUMillicores uint32 + RAMMB uint64 + StorageGB uint64 + Capabilities []byte +} + +// AnnounceOfferFor publishes or replaces provider's resource offer via +// pallet-resource-market's announce_offer_for, gated by AnnounceOrigin +// (EnsureRoot -- this bridge's sudo account) since the Provider Agent +// never talks to the chain directly (AGENTS.md). Idempotent: replacing an +// existing offer is a normal call, not an error, matching the pallet. +// +// Mirrors EnsureActive's submission shape exactly (serialize nonce use, +// sudo-wrap, sign, submit) but does not wait for finalization itself -- +// callers that need that guarantee should follow up with a +// FinalizedOffer read, the same separation ActivateProvider/ +// finalizedProvider already use for provider registration. +func (r *Registrar) AnnounceOfferFor(ctx context.Context, provider [32]byte, offer ResourceOffer) error { + r.mu.Lock() + defer r.mu.Unlock() + + version, err := r.rpc.RuntimeVersion(ctx) + if err != nil { + return err + } + genesisHex, err := r.rpc.BlockHash(ctx, 0) + if err != nil { + return err + } + genesis, err := fixedHash(genesisHex) + if err != nil { + return err + } + nonce, err := r.finalizedAccountNonce(ctx) + if err != nil { + return err + } + + inner := []byte{resourceMarketPalletIndex, announceOfferForCallIndex} + inner = append(inner, provider[:]...) + cpu := make([]byte, 4) + binary.LittleEndian.PutUint32(cpu, offer.CPUMillicores) + inner = append(inner, cpu...) + ram := make([]byte, 8) + binary.LittleEndian.PutUint64(ram, offer.RAMMB) + inner = append(inner, ram...) + storage := make([]byte, 8) + binary.LittleEndian.PutUint64(storage, offer.StorageGB) + inner = append(inner, storage...) + inner = append(inner, encodeBoundedBytes(offer.Capabilities)...) + + return r.submitSigned(ctx, append([]byte{sudoPalletIndex, sudoCallIndex}, inner...), nonce, version, genesis) +} + +// RemoveOfferFor withdraws provider's offer via remove_offer_for. Same +// origin/submission shape as AnnounceOfferFor. +func (r *Registrar) RemoveOfferFor(ctx context.Context, provider [32]byte) error { + r.mu.Lock() + defer r.mu.Unlock() + + version, err := r.rpc.RuntimeVersion(ctx) + if err != nil { + return err + } + genesisHex, err := r.rpc.BlockHash(ctx, 0) + if err != nil { + return err + } + genesis, err := fixedHash(genesisHex) + if err != nil { + return err + } + nonce, err := r.finalizedAccountNonce(ctx) + if err != nil { + return err + } + + inner := []byte{resourceMarketPalletIndex, removeOfferForCallIndex} + inner = append(inner, provider[:]...) + + return r.submitSigned(ctx, append([]byte{sudoPalletIndex, sudoCallIndex}, inner...), nonce, version, genesis) +} + +// FinalizedOffer reads pallet-resource-market's Offers map at blockHash +// for provider. found is false when no offer is published -- a normal +// state (never announced, or withdrawn), not a read failure. +func (c *RPCClient) FinalizedOffer(ctx context.Context, provider [32]byte, blockHash string) (ResourceOffer, bool, error) { + value, found, err := c.Storage(ctx, resourceMarketOfferStorageKey(provider), blockHash) + if err != nil { + return ResourceOffer{}, false, err + } + if !found { + return ResourceOffer{}, false, nil + } + offer, err := decodeResourceOffer(value) + if err != nil { + return ResourceOffer{}, false, err + } + return offer, true, nil +} + +func resourceMarketOfferStorageKey(provider [32]byte) string { + return mapStorageKey("ResourceMarket", "Offers", provider) +} + +// decodeResourceOffer decodes cpu(u32 LE) + ram(u64 LE) + storage(u64 LE) +// + capabilities(compact-length-prefixed bytes), in the field order +// pallet_resource_market::pallet::ResourceOffer declares them. +func decodeResourceOffer(data []byte) (ResourceOffer, error) { + if len(data) < 20 { + return ResourceOffer{}, errors.New("resource offer is shorter than its fixed-size fields") + } + offer := ResourceOffer{ + CPUMillicores: binary.LittleEndian.Uint32(data[0:4]), + RAMMB: binary.LittleEndian.Uint64(data[4:12]), + StorageGB: binary.LittleEndian.Uint64(data[12:20]), + } + length, offset, err := decodeCompactUint(data[20:]) + if err != nil { + return ResourceOffer{}, err + } + start := 20 + offset + end := start + int(length) + if end > len(data) { + return ResourceOffer{}, errors.New("resource offer capabilities length exceeds the encoded value") + } + offer.Capabilities = append([]byte(nil), data[start:end]...) + if end != len(data) { + return ResourceOffer{}, errors.New("resource offer has trailing bytes past its capabilities field") + } + return offer, nil +} + +// encodeBoundedBytes SCALE-encodes a BoundedVec/Vec: a compact +// length prefix followed by the raw bytes -- the same shape +// decodeResourceOffer's capabilities field decodes. +func encodeBoundedBytes(value []byte) []byte { + encoded := compactUint(uint64(len(value))) + return append(encoded, value...) +} diff --git a/control-plane/internal/blockchainbridge/resourcemarket_test.go b/control-plane/internal/blockchainbridge/resourcemarket_test.go new file mode 100644 index 0000000..7cfba30 --- /dev/null +++ b/control-plane/internal/blockchainbridge/resourcemarket_test.go @@ -0,0 +1,80 @@ +package blockchainbridge + +import ( + "encoding/binary" + "testing" +) + +func encodeResourceOfferForTest(o ResourceOffer) []byte { + data := make([]byte, 20) + binary.LittleEndian.PutUint32(data[0:4], o.CPUMillicores) + binary.LittleEndian.PutUint64(data[4:12], o.RAMMB) + binary.LittleEndian.PutUint64(data[12:20], o.StorageGB) + return append(data, encodeBoundedBytes(o.Capabilities)...) +} + +func TestDecodeResourceOfferRoundTripsFixedFieldsAndCapabilities(t *testing.T) { + cases := []ResourceOffer{ + {CPUMillicores: 2000, RAMMB: 4096, StorageGB: 100, Capabilities: []byte("gpu,fast-nvme")}, + {CPUMillicores: 1, RAMMB: 1, StorageGB: 1, Capabilities: nil}, + {CPUMillicores: 0, RAMMB: 0, StorageGB: 0, Capabilities: []byte{}}, + } + for _, want := range cases { + got, err := decodeResourceOffer(encodeResourceOfferForTest(want)) + if err != nil { + t.Fatalf("decodeResourceOffer(%+v): %v", want, err) + } + if got.CPUMillicores != want.CPUMillicores || got.RAMMB != want.RAMMB || got.StorageGB != want.StorageGB { + t.Fatalf("decodeResourceOffer() = %+v, want %+v", got, want) + } + if len(got.Capabilities) != len(want.Capabilities) { + t.Fatalf("capabilities length = %d, want %d", len(got.Capabilities), len(want.Capabilities)) + } + for i := range want.Capabilities { + if got.Capabilities[i] != want.Capabilities[i] { + t.Fatalf("capabilities mismatch at %d: got %v, want %v", i, got.Capabilities, want.Capabilities) + } + } + } +} + +func TestDecodeResourceOfferRejectsTruncatedAndTrailingData(t *testing.T) { + valid := encodeResourceOfferForTest(ResourceOffer{CPUMillicores: 1, RAMMB: 1, StorageGB: 1, Capabilities: []byte("x")}) + if _, err := decodeResourceOffer(valid[:19]); err == nil { + t.Fatal("expected an error decoding fewer than the fixed 20 bytes") + } + if _, err := decodeResourceOffer(append(valid, 0xFF)); err == nil { + t.Fatal("expected an error decoding a trailing byte past the declared capabilities length") + } + truncatedCapabilities := valid[:len(valid)-1] // claims 1 byte of capabilities but supplies 0 + if _, err := decodeResourceOffer(truncatedCapabilities); err == nil { + t.Fatal("expected an error when the capabilities length exceeds the remaining data") + } +} + +func TestEncodeBoundedBytesMatchesTheExistingCompactEncoder(t *testing.T) { + value := []byte("gpu,fast-nvme") + encoded := encodeBoundedBytes(value) + length, offset, err := decodeCompactUint(encoded) + if err != nil { + t.Fatalf("decodeCompactUint: %v", err) + } + if int(length) != len(value) { + t.Fatalf("encoded length prefix = %d, want %d", length, len(value)) + } + if string(encoded[offset:]) != string(value) { + t.Fatalf("encoded payload = %q, want %q", encoded[offset:], value) + } +} + +func TestResourceMarketOfferStorageKeyIsAPerProviderMapEntry(t *testing.T) { + var providerA, providerB [32]byte + providerA[0], providerB[0] = 1, 2 + keyA := resourceMarketOfferStorageKey(providerA) + if keyA == resourceMarketOfferStorageKey(providerB) { + t.Fatal("expected different providers to hash to different storage keys") + } + if keyA != resourceMarketOfferStorageKey(providerA) { + t.Fatal("expected a deterministic storage key for the same provider") + } +}