From f4e3cd42ead08417f6da66f864a39db9c08b0785 Mon Sep 17 00:00:00 2001 From: Joseph Chalabi Date: Mon, 27 Jul 2026 16:59:33 -0700 Subject: [PATCH 1/3] feat(inventory): persist committed snapshots Keep exact signed payloads pending until a successful chain post marks them public, and recover both pending and posted records across provider restarts. Signed-off-by: Joseph Chalabi --- tools/pconfig/bbolt/bbolt.go | 380 ++++++++++++++++++++++- tools/pconfig/bbolt/inventory_test.go | 51 +++ tools/pconfig/inventory_test.go | 231 ++++++++++++++ tools/pconfig/memory/memory.go | 193 +++++++++++- tools/pconfig/pconfig.go | 3 + verification/inventory/committed.go | 128 ++++++++ verification/inventory/committed_test.go | 140 +++++++++ 7 files changed, 1115 insertions(+), 11 deletions(-) create mode 100644 tools/pconfig/bbolt/inventory_test.go create mode 100644 tools/pconfig/inventory_test.go create mode 100644 verification/inventory/committed.go create mode 100644 verification/inventory/committed_test.go diff --git a/tools/pconfig/bbolt/bbolt.go b/tools/pconfig/bbolt/bbolt.go index 721a9a0b5..5d5e5eae7 100644 --- a/tools/pconfig/bbolt/bbolt.go +++ b/tools/pconfig/bbolt/bbolt.go @@ -1,6 +1,7 @@ package bbolt import ( + "bytes" "context" "crypto" "crypto/ecdsa" @@ -9,6 +10,8 @@ import ( "errors" "fmt" "math/big" + "sort" + "time" "go.etcd.io/bbolt" berrors "go.etcd.io/bbolt/errors" @@ -20,6 +23,7 @@ import ( ctypes "pkg.akt.dev/go/node/cert/v1" "github.com/akash-network/provider/tools/pconfig" + "github.com/akash-network/provider/verification/inventory" ) var ( @@ -28,15 +32,25 @@ var ( ErrDBClosed = errors.New("bbolt: database instance closed") ErrCertificateNotFoundInPEM = fmt.Errorf("%w: certificate not found in PEM", ctypes.ErrCertificate) ErrInvalidPubKeyType = errors.New("bbolt: invalid pubkey type") + errInvalidSnapshotRecord = errors.New("bbolt: invalid committed inventory snapshot record") ) var ( - bucketAccounts = []byte("accounts") - bucketBidEngine = []byte("bidengine") - keyPubKey = []byte("pubkey") - bucketCertificates = []byte("certificates") - keyCertificate = []byte("certificate") - keyNextKey = []byte("nextkey") + bucketAccounts = []byte("accounts") + bucketBidEngine = []byte("bidengine") + bucketInventorySnapshots = []byte("inventory-snapshots") + bucketInventorySnapshotRecords = []byte("records") + keyPubKey = []byte("pubkey") + bucketCertificates = []byte("certificates") + keyCertificate = []byte("certificate") + keyNextKey = []byte("nextkey") + keyInventorySnapshotHash = []byte("hash") + keyInventorySnapshotPayload = []byte("payload") + keyInventorySnapshotSignature = []byte("signature") + keyInventorySnapshotProvider = []byte("provider") + keyInventorySnapshotState = []byte("state") + keyInventorySnapshotPostedAt = []byte("posted-at") + keyLatestInventorySnapshot = []byte("latest-posted") ) type impl struct { @@ -63,6 +77,16 @@ func NewBBolt(path string) (pconfig.Storage, error) { return fmt.Errorf("create bucket: %s", err) } + snapshots, err := tx.CreateBucketIfNotExists(bucketInventorySnapshots) + if err != nil { + return fmt.Errorf("create bucket: %s", err) + } + + _, err = snapshots.CreateBucketIfNotExists(bucketInventorySnapshotRecords) + if err != nil { + return fmt.Errorf("create bucket: %s", err) + } + return nil }) if err != nil { @@ -81,6 +105,10 @@ func (b *impl) BidEngine() pconfig.BidEngine { return b } +func (b *impl) InventorySnapshots() inventory.CommittedSnapshotStore { + return b +} + func (b *impl) Close() error { select { case <-b.closed: @@ -445,3 +473,343 @@ func (b *impl) GetOrdersNextKey(_ context.Context) ([]byte, error) { return res, nil } + +func (b *impl) Stage(_ context.Context, snapshot inventory.Snapshot) error { + if b.isClosed() { + return ErrDBClosed + } + + record, err := inventory.NewPendingCommittedSnapshot(snapshot) + if err != nil { + return err + } + + return b.db.Update(func(tx *bbolt.Tx) error { + records, err := inventorySnapshotRecordsBucket(tx) + if err != nil { + return err + } + + recordBucket := records.Bucket(record.Snapshot.Hash) + if recordBucket != nil { + current, err := readInventorySnapshotRecord(recordBucket, record.Snapshot.Hash) + if err != nil { + return err + } + if !inventorySnapshotsEqual(current.Snapshot, record.Snapshot) { + return inventory.ErrCommittedSnapshotConflict + } + + return nil + } + + recordBucket, err = records.CreateBucket(record.Snapshot.Hash) + if err != nil { + return err + } + + return writeInventorySnapshotRecord(recordBucket, record) + }) +} + +func (b *impl) Pending(_ context.Context) ([]inventory.CommittedSnapshot, error) { + if b.isClosed() { + return nil, ErrDBClosed + } + + records := make([]inventory.CommittedSnapshot, 0) + err := b.db.View(func(tx *bbolt.Tx) error { + recordsBucket, err := inventorySnapshotRecordsBucket(tx) + if err != nil { + return err + } + + return recordsBucket.ForEachBucket(func(hash []byte) error { + record, err := readInventorySnapshotRecord(recordsBucket.Bucket(hash), hash) + if err != nil { + return err + } + if record.State == inventory.CommittedSnapshotStatePending { + records = append(records, record) + } + + return nil + }) + }) + if err != nil { + return nil, err + } + + sort.Slice(records, func(lhs, rhs int) bool { + return bytes.Compare(records[lhs].Snapshot.Hash, records[rhs].Snapshot.Hash) < 0 + }) + + return records, nil +} + +func (b *impl) MarkPosted(_ context.Context, hash []byte, postedAt time.Time) error { + if b.isClosed() { + return ErrDBClosed + } + if err := inventory.ValidateCommittedSnapshotHash(hash); err != nil { + return err + } + if postedAt.IsZero() { + return pconfig.ErrInvalidArgs + } + postedAt = postedAt.Round(0).UTC() + + return b.db.Update(func(tx *bbolt.Tx) error { + snapshots := tx.Bucket(bucketInventorySnapshots) + if snapshots == nil { + return ErrUninitialized + } + records := snapshots.Bucket(bucketInventorySnapshotRecords) + if records == nil { + return ErrUninitialized + } + + recordBucket := records.Bucket(hash) + if recordBucket == nil { + return inventory.ErrCommittedSnapshotNotFound + } + + record, err := readInventorySnapshotRecord(recordBucket, hash) + if err != nil { + return err + } + if record.State == inventory.CommittedSnapshotStatePosted { + if record.PostedAt.Equal(postedAt) { + return nil + } + + return inventory.ErrCommittedSnapshotConflict + } + + shouldUpdateLatest := true + latestHash := snapshots.Get(keyLatestInventorySnapshot) + if latestHash != nil { + if err := inventory.ValidateCommittedSnapshotHash(latestHash); err != nil { + return fmt.Errorf("%w: latest hash: %s", errInvalidSnapshotRecord, err) + } + + latestBucket := records.Bucket(latestHash) + if latestBucket == nil { + return fmt.Errorf("%w: latest snapshot does not exist", errInvalidSnapshotRecord) + } + + latest, err := readInventorySnapshotRecord(latestBucket, latestHash) + if err != nil { + return err + } + if latest.State != inventory.CommittedSnapshotStatePosted { + return fmt.Errorf("%w: latest snapshot is not posted", errInvalidSnapshotRecord) + } + + shouldUpdateLatest = !postedAt.Before(latest.PostedAt) + } + + record, err = inventory.MarkCommittedSnapshotPosted(record, postedAt) + if err != nil { + return err + } + if err := writeInventorySnapshotRecord(recordBucket, record); err != nil { + return err + } + if shouldUpdateLatest { + return snapshots.Put(keyLatestInventorySnapshot, record.Snapshot.Hash) + } + + return nil + }) +} + +func (b *impl) Get(_ context.Context, hash []byte) (inventory.CommittedSnapshot, error) { + if b.isClosed() { + return inventory.CommittedSnapshot{}, ErrDBClosed + } + if err := inventory.ValidateCommittedSnapshotHash(hash); err != nil { + return inventory.CommittedSnapshot{}, err + } + + var record inventory.CommittedSnapshot + err := b.db.View(func(tx *bbolt.Tx) error { + records, err := inventorySnapshotRecordsBucket(tx) + if err != nil { + return err + } + + recordBucket := records.Bucket(hash) + if recordBucket == nil { + return inventory.ErrCommittedSnapshotNotFound + } + + record, err = readInventorySnapshotRecord(recordBucket, hash) + if err != nil { + return err + } + if record.State != inventory.CommittedSnapshotStatePosted { + return inventory.ErrCommittedSnapshotNotFound + } + + return nil + }) + if err != nil { + return inventory.CommittedSnapshot{}, err + } + + return record, nil +} + +func (b *impl) Latest(_ context.Context) (inventory.CommittedSnapshot, error) { + if b.isClosed() { + return inventory.CommittedSnapshot{}, ErrDBClosed + } + + var record inventory.CommittedSnapshot + err := b.db.View(func(tx *bbolt.Tx) error { + snapshots := tx.Bucket(bucketInventorySnapshots) + if snapshots == nil { + return ErrUninitialized + } + + hash := snapshots.Get(keyLatestInventorySnapshot) + if hash == nil { + return inventory.ErrCommittedSnapshotNotFound + } + if err := inventory.ValidateCommittedSnapshotHash(hash); err != nil { + return fmt.Errorf("%w: latest hash: %s", errInvalidSnapshotRecord, err) + } + + records := snapshots.Bucket(bucketInventorySnapshotRecords) + if records == nil { + return ErrUninitialized + } + recordBucket := records.Bucket(hash) + if recordBucket == nil { + return fmt.Errorf("%w: latest snapshot does not exist", errInvalidSnapshotRecord) + } + + var err error + record, err = readInventorySnapshotRecord(recordBucket, hash) + if err != nil { + return err + } + if record.State != inventory.CommittedSnapshotStatePosted { + return fmt.Errorf("%w: latest snapshot is not posted", errInvalidSnapshotRecord) + } + + return nil + }) + if err != nil { + return inventory.CommittedSnapshot{}, err + } + + return record, nil +} + +func (b *impl) isClosed() bool { + select { + case <-b.closed: + return true + default: + return false + } +} + +func inventorySnapshotRecordsBucket(tx *bbolt.Tx) (*bbolt.Bucket, error) { + snapshots := tx.Bucket(bucketInventorySnapshots) + if snapshots == nil { + return nil, ErrUninitialized + } + + records := snapshots.Bucket(bucketInventorySnapshotRecords) + if records == nil { + return nil, ErrUninitialized + } + + return records, nil +} + +func writeInventorySnapshotRecord(bucket *bbolt.Bucket, record inventory.CommittedSnapshot) error { + if err := inventory.ValidateCommittedSnapshot(record); err != nil { + return err + } + + if err := bucket.Put(keyInventorySnapshotHash, record.Snapshot.Hash); err != nil { + return err + } + if err := bucket.Put(keyInventorySnapshotPayload, record.Snapshot.Payload); err != nil { + return err + } + if err := bucket.Put(keyInventorySnapshotSignature, record.Snapshot.Signature); err != nil { + return err + } + if err := bucket.Put(keyInventorySnapshotProvider, []byte(record.Snapshot.Provider)); err != nil { + return err + } + if err := bucket.Put(keyInventorySnapshotState, []byte{byte(record.State)}); err != nil { + return err + } + + if record.PostedAt.IsZero() { + return bucket.Delete(keyInventorySnapshotPostedAt) + } + + postedAt, err := record.PostedAt.MarshalBinary() + if err != nil { + return err + } + + return bucket.Put(keyInventorySnapshotPostedAt, postedAt) +} + +func readInventorySnapshotRecord(bucket *bbolt.Bucket, hash []byte) (inventory.CommittedSnapshot, error) { + if bucket == nil { + return inventory.CommittedSnapshot{}, fmt.Errorf("%w: missing record bucket", errInvalidSnapshotRecord) + } + if err := inventory.ValidateCommittedSnapshotHash(hash); err != nil { + return inventory.CommittedSnapshot{}, fmt.Errorf("%w: bucket hash: %s", errInvalidSnapshotRecord, err) + } + + storedHash := bucket.Get(keyInventorySnapshotHash) + if !bytes.Equal(storedHash, hash) { + return inventory.CommittedSnapshot{}, fmt.Errorf("%w: stored hash does not match record key", errInvalidSnapshotRecord) + } + + state := bucket.Get(keyInventorySnapshotState) + if len(state) != 1 { + return inventory.CommittedSnapshot{}, fmt.Errorf("%w: state must be one byte", errInvalidSnapshotRecord) + } + + record := inventory.CommittedSnapshot{ + Snapshot: inventory.Snapshot{ + Payload: append([]byte(nil), bucket.Get(keyInventorySnapshotPayload)...), + Hash: append([]byte(nil), storedHash...), + Signature: append([]byte(nil), bucket.Get(keyInventorySnapshotSignature)...), + Provider: string(bucket.Get(keyInventorySnapshotProvider)), + }, + State: inventory.CommittedSnapshotState(state[0]), + } + + postedAt := bucket.Get(keyInventorySnapshotPostedAt) + if len(postedAt) != 0 { + if err := record.PostedAt.UnmarshalBinary(postedAt); err != nil { + return inventory.CommittedSnapshot{}, fmt.Errorf("%w: posted at: %s", errInvalidSnapshotRecord, err) + } + record.PostedAt = record.PostedAt.Round(0).UTC() + } + + if err := inventory.ValidateCommittedSnapshot(record); err != nil { + return inventory.CommittedSnapshot{}, fmt.Errorf("%w: %s", errInvalidSnapshotRecord, err) + } + + return record, nil +} + +func inventorySnapshotsEqual(lhs, rhs inventory.Snapshot) bool { + return bytes.Equal(lhs.Payload, rhs.Payload) && + bytes.Equal(lhs.Hash, rhs.Hash) && + bytes.Equal(lhs.Signature, rhs.Signature) && + lhs.Provider == rhs.Provider +} diff --git a/tools/pconfig/bbolt/inventory_test.go b/tools/pconfig/bbolt/inventory_test.go new file mode 100644 index 000000000..7682c967a --- /dev/null +++ b/tools/pconfig/bbolt/inventory_test.go @@ -0,0 +1,51 @@ +package bbolt + +import ( + "context" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" + boltdb "go.etcd.io/bbolt" + + "github.com/akash-network/provider/verification/inventory" +) + +func TestInventorySnapshotsRejectCorruptRecord(t *testing.T) { + ctx := context.Background() + payload := []byte("payload") + snapshot := inventory.Snapshot{ + Payload: payload, + Hash: inventory.HashPayload(payload), + Signature: []byte("signature"), + Provider: "akash1provider", + } + + storage, err := NewBBolt(filepath.Join(t.TempDir(), "pconfig.db")) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, storage.Close()) + }) + + store := storage.InventorySnapshots() + require.NoError(t, store.Stage(ctx, snapshot)) + require.NoError(t, store.MarkPosted(ctx, snapshot.Hash, time.Now().UTC())) + + db := storage.(*impl).db + err = db.Update(func(tx *boltdb.Tx) error { + records, err := inventorySnapshotRecordsBucket(tx) + if err != nil { + return err + } + + return records.Bucket(snapshot.Hash).Put(keyInventorySnapshotPayload, []byte("corrupt")) + }) + require.NoError(t, err) + + _, err = store.Get(ctx, snapshot.Hash) + require.ErrorIs(t, err, errInvalidSnapshotRecord) + + _, err = store.Latest(ctx) + require.ErrorIs(t, err, errInvalidSnapshotRecord) +} diff --git a/tools/pconfig/inventory_test.go b/tools/pconfig/inventory_test.go new file mode 100644 index 000000000..0e4222789 --- /dev/null +++ b/tools/pconfig/inventory_test.go @@ -0,0 +1,231 @@ +package pconfig_test + +import ( + "context" + "errors" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/akash-network/provider/tools/pconfig/bbolt" + "github.com/akash-network/provider/verification/inventory" +) + +func newInventorySnapshot(payload string) inventory.Snapshot { + data := []byte(payload) + + return inventory.Snapshot{ + Payload: data, + Hash: inventory.HashPayload(data), + Signature: []byte("signature-" + payload), + Provider: "akash1provider", + } +} + +func TestInventorySnapshotsLifecycle(t *testing.T) { + dbs := initTestBackends(t) + defer func() { + for _, db := range dbs { + db.cleanup() + } + }() + + for _, db := range dbs { + t.Run(db.name, func(t *testing.T) { + ctx := context.Background() + store := db.InventorySnapshots() + snapshot := newInventorySnapshot("first") + original := inventory.CloneSnapshot(snapshot) + + err := store.Stage(ctx, snapshot) + require.NoError(t, err) + + snapshot.Payload[0] = 'x' + snapshot.Hash[0] = 0 + snapshot.Signature[0] = 'x' + + _, err = store.Get(ctx, original.Hash) + require.ErrorIs(t, err, inventory.ErrCommittedSnapshotNotFound) + + _, err = store.Latest(ctx) + require.ErrorIs(t, err, inventory.ErrCommittedSnapshotNotFound) + + pending, err := store.Pending(ctx) + require.NoError(t, err) + require.Equal(t, []inventory.CommittedSnapshot{{ + Snapshot: original, + State: inventory.CommittedSnapshotStatePending, + }}, pending) + + pending[0].Snapshot.Payload[0] = 'x' + pending, err = store.Pending(ctx) + require.NoError(t, err) + require.Equal(t, []byte("first"), pending[0].Snapshot.Payload) + + postedAt := time.Date(2026, time.July, 27, 19, 34, 56, 789, time.UTC) + err = store.MarkPosted(ctx, original.Hash, postedAt) + require.NoError(t, err) + + expected := inventory.CommittedSnapshot{ + Snapshot: original, + State: inventory.CommittedSnapshotStatePosted, + PostedAt: postedAt, + } + + record, err := store.Get(ctx, original.Hash) + require.NoError(t, err) + require.Equal(t, expected, record) + + latest, err := store.Latest(ctx) + require.NoError(t, err) + require.Equal(t, expected, latest) + + pending, err = store.Pending(ctx) + require.NoError(t, err) + require.Empty(t, pending) + + record.Snapshot.Signature[0] = 'x' + record, err = store.Get(ctx, original.Hash) + require.NoError(t, err) + require.Equal(t, expected, record) + + err = store.Stage(ctx, original) + require.NoError(t, err) + + record, err = store.Get(ctx, original.Hash) + require.NoError(t, err) + require.Equal(t, expected, record) + }) + } +} + +func TestInventorySnapshotsLatestUsesSuccessfulPostTime(t *testing.T) { + dbs := initTestBackends(t) + defer func() { + for _, db := range dbs { + db.cleanup() + } + }() + + for _, db := range dbs { + t.Run(db.name, func(t *testing.T) { + ctx := context.Background() + store := db.InventorySnapshots() + older := newInventorySnapshot("older") + newer := newInventorySnapshot("newer") + olderPostedAt := time.Date(2026, time.July, 27, 10, 0, 0, 0, time.UTC) + newerPostedAt := olderPostedAt.Add(time.Minute) + + require.NoError(t, store.Stage(ctx, older)) + require.NoError(t, store.Stage(ctx, newer)) + require.NoError(t, store.MarkPosted(ctx, newer.Hash, newerPostedAt)) + require.NoError(t, store.MarkPosted(ctx, older.Hash, olderPostedAt)) + + latest, err := store.Latest(ctx) + require.NoError(t, err) + require.Equal(t, newer.Hash, latest.Snapshot.Hash) + require.Equal(t, newerPostedAt, latest.PostedAt) + + olderRecord, err := store.Get(ctx, older.Hash) + require.NoError(t, err) + require.Equal(t, olderPostedAt, olderRecord.PostedAt) + }) + } +} + +func TestInventorySnapshotsRejectInvalidLifecycle(t *testing.T) { + dbs := initTestBackends(t) + defer func() { + for _, db := range dbs { + db.cleanup() + } + }() + + for _, db := range dbs { + t.Run(db.name, func(t *testing.T) { + ctx := context.Background() + store := db.InventorySnapshots() + snapshot := newInventorySnapshot("payload") + invalid := inventory.CloneSnapshot(snapshot) + invalid.Hash = inventory.HashPayload([]byte("different")) + + require.Error(t, store.Stage(ctx, invalid)) + require.Error(t, store.MarkPosted(ctx, snapshot.Hash, time.Time{})) + require.ErrorIs( + t, + store.MarkPosted(ctx, snapshot.Hash, time.Now().UTC()), + inventory.ErrCommittedSnapshotNotFound, + ) + + require.NoError(t, store.Stage(ctx, snapshot)) + postedAt := time.Now().UTC() + require.NoError(t, store.MarkPosted(ctx, snapshot.Hash, postedAt)) + require.NoError(t, store.MarkPosted(ctx, snapshot.Hash, postedAt)) + require.ErrorIs( + t, + store.MarkPosted(ctx, snapshot.Hash, postedAt.Add(time.Second)), + inventory.ErrCommittedSnapshotConflict, + ) + + conflicting := inventory.CloneSnapshot(snapshot) + conflicting.Signature = []byte("different signature") + require.ErrorIs(t, store.Stage(ctx, conflicting), inventory.ErrCommittedSnapshotConflict) + + _, err := store.Get(ctx, []byte("short")) + require.Error(t, err) + require.False(t, errors.Is(err, inventory.ErrCommittedSnapshotNotFound)) + }) + } +} + +func TestBBoltInventorySnapshotsPersistAcrossRestart(t *testing.T) { + ctx := context.Background() + dbPath := filepath.Join(t.TempDir(), "pconfig.db") + postedSnapshot := newInventorySnapshot("posted") + pendingSnapshot := newInventorySnapshot("pending") + postedAt := time.Date(2026, time.July, 27, 19, 34, 56, 123456789, time.UTC) + + db, err := bbolt.NewBBolt(dbPath) + require.NoError(t, err) + store := db.InventorySnapshots() + require.NoError(t, store.Stage(ctx, postedSnapshot)) + require.NoError(t, store.Stage(ctx, pendingSnapshot)) + require.NoError(t, store.MarkPosted(ctx, postedSnapshot.Hash, postedAt)) + require.NoError(t, db.Close()) + + db, err = bbolt.NewBBolt(dbPath) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, db.Close()) + }) + store = db.InventorySnapshots() + + record, err := store.Get(ctx, postedSnapshot.Hash) + require.NoError(t, err) + require.Equal(t, postedSnapshot.Payload, record.Snapshot.Payload) + require.Equal(t, postedSnapshot.Signature, record.Snapshot.Signature) + require.Equal(t, postedSnapshot.Hash, record.Snapshot.Hash) + require.Equal(t, postedSnapshot.Provider, record.Snapshot.Provider) + require.Equal(t, postedAt, record.PostedAt) + require.Equal(t, inventory.CommittedSnapshotStatePosted, record.State) + + latest, err := store.Latest(ctx) + require.NoError(t, err) + require.Equal(t, record, latest) + + _, err = store.Get(ctx, pendingSnapshot.Hash) + require.ErrorIs(t, err, inventory.ErrCommittedSnapshotNotFound) + + pending, err := store.Pending(ctx) + require.NoError(t, err) + require.Equal(t, pendingSnapshot, pending[0].Snapshot) + + pendingPostedAt := postedAt.Add(time.Minute) + require.NoError(t, store.MarkPosted(ctx, pendingSnapshot.Hash, pendingPostedAt)) + latest, err = store.Latest(ctx) + require.NoError(t, err) + require.Equal(t, pendingSnapshot, latest.Snapshot) + require.Equal(t, pendingPostedAt, latest.PostedAt) +} diff --git a/tools/pconfig/memory/memory.go b/tools/pconfig/memory/memory.go index 8968db0f5..4ef3e671c 100644 --- a/tools/pconfig/memory/memory.go +++ b/tools/pconfig/memory/memory.go @@ -1,16 +1,21 @@ package memory import ( + "bytes" "context" "crypto" "crypto/x509" + "errors" "math/big" + "sort" "sync" + "time" cryptotypes "github.com/cosmos/cosmos-sdk/crypto/types" sdk "github.com/cosmos/cosmos-sdk/types" "github.com/akash-network/provider/tools/pconfig" + "github.com/akash-network/provider/verification/inventory" ) type certificate struct { @@ -30,17 +35,21 @@ type bidengine struct { type impl struct { closed chan struct{} - lock sync.RWMutex - accounts map[string]account - bidengine bidengine + lock sync.RWMutex + accounts map[string]account + bidengine bidengine + inventorySnapshots map[[inventory.SnapshotHashSize]byte]inventory.CommittedSnapshot + latestSnapshot [inventory.SnapshotHashSize]byte + hasLatestSnapshot bool } var _ pconfig.Storage = (*impl)(nil) func NewMemory() (pconfig.Storage, error) { b := &impl{ - closed: make(chan struct{}), - accounts: make(map[string]account), + closed: make(chan struct{}), + accounts: make(map[string]account), + inventorySnapshots: make(map[[inventory.SnapshotHashSize]byte]inventory.CommittedSnapshot), } return b, nil @@ -50,6 +59,10 @@ func (i *impl) BidEngine() pconfig.BidEngine { return i } +func (i *impl) InventorySnapshots() inventory.CommittedSnapshotStore { + return i +} + func (i *impl) AddAccount(_ context.Context, address sdk.Address, pubkey cryptotypes.PubKey) error { if address.Empty() || pubkey == nil { return pconfig.ErrInvalidArgs @@ -216,3 +229,173 @@ func (i *impl) GetOrdersNextKey(_ context.Context) ([]byte, error) { return res, nil } + +func (i *impl) Stage(_ context.Context, snapshot inventory.Snapshot) error { + record, err := inventory.NewPendingCommittedSnapshot(snapshot) + if err != nil { + return err + } + + key, err := inventorySnapshotKey(record.Snapshot.Hash) + if err != nil { + return err + } + + defer i.lock.Unlock() + i.lock.Lock() + + if current, exists := i.inventorySnapshots[key]; exists { + if err := inventory.ValidateCommittedSnapshot(current); err != nil { + return err + } + if !inventorySnapshotsEqual(current.Snapshot, record.Snapshot) { + return inventory.ErrCommittedSnapshotConflict + } + + return nil + } + + i.inventorySnapshots[key] = inventory.CloneCommittedSnapshot(record) + + return nil +} + +func (i *impl) Pending(_ context.Context) ([]inventory.CommittedSnapshot, error) { + defer i.lock.RUnlock() + i.lock.RLock() + + records := make([]inventory.CommittedSnapshot, 0) + for _, record := range i.inventorySnapshots { + if err := inventory.ValidateCommittedSnapshot(record); err != nil { + return nil, err + } + if record.State == inventory.CommittedSnapshotStatePending { + records = append(records, inventory.CloneCommittedSnapshot(record)) + } + } + + sort.Slice(records, func(lhs, rhs int) bool { + return bytes.Compare(records[lhs].Snapshot.Hash, records[rhs].Snapshot.Hash) < 0 + }) + + return records, nil +} + +func (i *impl) MarkPosted(_ context.Context, hash []byte, postedAt time.Time) error { + if postedAt.IsZero() { + return pconfig.ErrInvalidArgs + } + + key, err := inventorySnapshotKey(hash) + if err != nil { + return err + } + postedAt = postedAt.Round(0).UTC() + + defer i.lock.Unlock() + i.lock.Lock() + + record, exists := i.inventorySnapshots[key] + if !exists { + return inventory.ErrCommittedSnapshotNotFound + } + if err := inventory.ValidateCommittedSnapshot(record); err != nil { + return err + } + if record.State == inventory.CommittedSnapshotStatePosted { + if record.PostedAt.Equal(postedAt) { + return nil + } + + return inventory.ErrCommittedSnapshotConflict + } + + shouldUpdateLatest := !i.hasLatestSnapshot + if i.hasLatestSnapshot { + latest, exists := i.inventorySnapshots[i.latestSnapshot] + if !exists { + return errors.New("memory: latest committed inventory snapshot does not exist") + } + if err := inventory.ValidateCommittedSnapshot(latest); err != nil { + return err + } + if latest.State != inventory.CommittedSnapshotStatePosted { + return errors.New("memory: latest committed inventory snapshot is not posted") + } + + shouldUpdateLatest = !postedAt.Before(latest.PostedAt) + } + + record, err = inventory.MarkCommittedSnapshotPosted(record, postedAt) + if err != nil { + return err + } + i.inventorySnapshots[key] = inventory.CloneCommittedSnapshot(record) + if shouldUpdateLatest { + i.latestSnapshot = key + i.hasLatestSnapshot = true + } + + return nil +} + +func (i *impl) Get(_ context.Context, hash []byte) (inventory.CommittedSnapshot, error) { + key, err := inventorySnapshotKey(hash) + if err != nil { + return inventory.CommittedSnapshot{}, err + } + + defer i.lock.RUnlock() + i.lock.RLock() + + record, exists := i.inventorySnapshots[key] + if !exists || record.State != inventory.CommittedSnapshotStatePosted { + return inventory.CommittedSnapshot{}, inventory.ErrCommittedSnapshotNotFound + } + if err := inventory.ValidateCommittedSnapshot(record); err != nil { + return inventory.CommittedSnapshot{}, err + } + + return inventory.CloneCommittedSnapshot(record), nil +} + +func (i *impl) Latest(_ context.Context) (inventory.CommittedSnapshot, error) { + defer i.lock.RUnlock() + i.lock.RLock() + + if !i.hasLatestSnapshot { + return inventory.CommittedSnapshot{}, inventory.ErrCommittedSnapshotNotFound + } + + record, exists := i.inventorySnapshots[i.latestSnapshot] + if !exists { + return inventory.CommittedSnapshot{}, errors.New("memory: latest committed inventory snapshot does not exist") + } + if err := inventory.ValidateCommittedSnapshot(record); err != nil { + return inventory.CommittedSnapshot{}, err + } + if record.State != inventory.CommittedSnapshotStatePosted { + return inventory.CommittedSnapshot{}, errors.New("memory: latest committed inventory snapshot is not posted") + } + + return inventory.CloneCommittedSnapshot(record), nil +} + +func inventorySnapshotKey(hash []byte) ([inventory.SnapshotHashSize]byte, error) { + var key [inventory.SnapshotHashSize]byte + + if err := inventory.ValidateCommittedSnapshotHash(hash); err != nil { + return key, err + } + + copy(key[:], hash) + + return key, nil +} + +func inventorySnapshotsEqual(lhs, rhs inventory.Snapshot) bool { + return bytes.Equal(lhs.Payload, rhs.Payload) && + bytes.Equal(lhs.Hash, rhs.Hash) && + bytes.Equal(lhs.Signature, rhs.Signature) && + lhs.Provider == rhs.Provider +} diff --git a/tools/pconfig/pconfig.go b/tools/pconfig/pconfig.go index d9e642a35..061ee478c 100644 --- a/tools/pconfig/pconfig.go +++ b/tools/pconfig/pconfig.go @@ -10,6 +10,8 @@ import ( cryptotypes "github.com/cosmos/cosmos-sdk/crypto/types" sdk "github.com/cosmos/cosmos-sdk/types" + + "github.com/akash-network/provider/verification/inventory" ) var ( @@ -44,5 +46,6 @@ type Storage interface { StorageR StorageW BidEngine() BidEngine + InventorySnapshots() inventory.CommittedSnapshotStore Close() error } diff --git a/verification/inventory/committed.go b/verification/inventory/committed.go new file mode 100644 index 000000000..0247c0bfa --- /dev/null +++ b/verification/inventory/committed.go @@ -0,0 +1,128 @@ +package inventory + +import ( + "bytes" + "context" + "crypto/sha256" + "errors" + "fmt" + "time" +) + +const SnapshotHashSize = sha256.Size + +var ( + ErrCommittedSnapshotNotFound = errors.New("inventory: committed snapshot not found") + ErrCommittedSnapshotConflict = errors.New("inventory: committed snapshot conflict") + + errInvalidCommittedSnapshotHash = errors.New("invalid committed inventory snapshot hash") + errInvalidCommittedSnapshotState = errors.New("invalid committed inventory snapshot state") + errInvalidCommittedSnapshotPostedAt = errors.New("invalid committed inventory snapshot post time") +) + +type CommittedSnapshotState uint8 + +const ( + CommittedSnapshotStatePending CommittedSnapshotState = iota + 1 + CommittedSnapshotStatePosted +) + +type CommittedSnapshot struct { + Snapshot Snapshot + State CommittedSnapshotState + PostedAt time.Time +} + +type CommittedSnapshotReader interface { + Get(context.Context, []byte) (CommittedSnapshot, error) + Latest(context.Context) (CommittedSnapshot, error) +} + +type CommittedSnapshotStore interface { + CommittedSnapshotReader + Stage(context.Context, Snapshot) error + Pending(context.Context) ([]CommittedSnapshot, error) + MarkPosted(context.Context, []byte, time.Time) error +} + +func ValidateCommittedSnapshotHash(hash []byte) error { + if len(hash) != SnapshotHashSize { + return fmt.Errorf("%w: expected %d bytes, got %d", errInvalidCommittedSnapshotHash, SnapshotHashSize, len(hash)) + } + + return nil +} + +func NewPendingCommittedSnapshot(snapshot Snapshot) (CommittedSnapshot, error) { + record := CommittedSnapshot{ + Snapshot: CloneSnapshot(snapshot), + State: CommittedSnapshotStatePending, + } + if err := ValidateCommittedSnapshot(record); err != nil { + return CommittedSnapshot{}, err + } + + return record, nil +} + +func MarkCommittedSnapshotPosted(record CommittedSnapshot, postedAt time.Time) (CommittedSnapshot, error) { + if err := ValidateCommittedSnapshot(record); err != nil { + return CommittedSnapshot{}, err + } + if record.State != CommittedSnapshotStatePending { + return CommittedSnapshot{}, ErrCommittedSnapshotConflict + } + if postedAt.IsZero() { + return CommittedSnapshot{}, errInvalidCommittedSnapshotPostedAt + } + + record = CloneCommittedSnapshot(record) + record.State = CommittedSnapshotStatePosted + record.PostedAt = postedAt.Round(0).UTC() + + return record, nil +} + +func ValidateCommittedSnapshot(record CommittedSnapshot) error { + if err := ValidateSnapshot(&record.Snapshot); err != nil { + return err + } + if err := ValidateCommittedSnapshotHash(record.Snapshot.Hash); err != nil { + return err + } + if !bytes.Equal(record.Snapshot.Hash, HashPayload(record.Snapshot.Payload)) { + return errInvalidCommittedSnapshotHash + } + + switch record.State { + case CommittedSnapshotStatePending: + if !record.PostedAt.IsZero() { + return errInvalidCommittedSnapshotPostedAt + } + case CommittedSnapshotStatePosted: + if record.PostedAt.IsZero() { + return errInvalidCommittedSnapshotPostedAt + } + default: + return errInvalidCommittedSnapshotState + } + + return nil +} + +func CloneSnapshot(snapshot Snapshot) Snapshot { + return Snapshot{ + Payload: append([]byte(nil), snapshot.Payload...), + Hash: append([]byte(nil), snapshot.Hash...), + Signature: append([]byte(nil), snapshot.Signature...), + Provider: snapshot.Provider, + } +} + +func CloneCommittedSnapshot(record CommittedSnapshot) CommittedSnapshot { + return CommittedSnapshot{ + Snapshot: CloneSnapshot(record.Snapshot), + State: record.State, + PostedAt: record.PostedAt, + } +} diff --git a/verification/inventory/committed_test.go b/verification/inventory/committed_test.go new file mode 100644 index 000000000..5533e1dd4 --- /dev/null +++ b/verification/inventory/committed_test.go @@ -0,0 +1,140 @@ +package inventory + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func testCommittedSnapshot() Snapshot { + payload := "payload" + data := []byte(payload) + + return Snapshot{ + Payload: data, + Hash: HashPayload(data), + Signature: []byte("signature-" + payload), + Provider: "akash1provider", + } +} + +func TestNewPendingCommittedSnapshot(t *testing.T) { + snapshot := testCommittedSnapshot() + + record, err := NewPendingCommittedSnapshot(snapshot) + require.NoError(t, err) + require.Equal(t, CommittedSnapshotStatePending, record.State) + require.True(t, record.PostedAt.IsZero()) + require.Equal(t, snapshot, record.Snapshot) + + snapshot.Payload[0] = 'x' + snapshot.Hash[0] = 0 + snapshot.Signature[0] = 'x' + + require.Equal(t, []byte("payload"), record.Snapshot.Payload) + require.Equal(t, HashPayload([]byte("payload")), record.Snapshot.Hash) + require.Equal(t, []byte("signature-payload"), record.Snapshot.Signature) +} + +func TestNewPendingCommittedSnapshotRejectsInvalidHash(t *testing.T) { + snapshot := testCommittedSnapshot() + snapshot.Hash = HashPayload([]byte("different payload")) + + record, err := NewPendingCommittedSnapshot(snapshot) + require.Error(t, err) + require.Empty(t, record) +} + +func TestMarkCommittedSnapshotPosted(t *testing.T) { + record, err := NewPendingCommittedSnapshot(testCommittedSnapshot()) + require.NoError(t, err) + + postedAt := time.Date(2026, time.July, 27, 12, 34, 56, 789, time.FixedZone("test", -7*60*60)) + posted, err := MarkCommittedSnapshotPosted(record, postedAt) + require.NoError(t, err) + + require.Equal(t, CommittedSnapshotStatePosted, posted.State) + require.Equal(t, postedAt.UTC(), posted.PostedAt) + require.Equal(t, record.Snapshot, posted.Snapshot) + require.True(t, record.PostedAt.IsZero()) +} + +func TestMarkCommittedSnapshotPostedRejectsInvalidTransition(t *testing.T) { + record, err := NewPendingCommittedSnapshot(testCommittedSnapshot()) + require.NoError(t, err) + + posted, err := MarkCommittedSnapshotPosted(record, time.Time{}) + require.Error(t, err) + require.Empty(t, posted) + + record.State = CommittedSnapshotStatePosted + record.PostedAt = time.Now().UTC() + + posted, err = MarkCommittedSnapshotPosted(record, time.Now()) + require.ErrorIs(t, err, ErrCommittedSnapshotConflict) + require.Empty(t, posted) +} + +func TestValidateCommittedSnapshotState(t *testing.T) { + snapshot := testCommittedSnapshot() + postedAt := time.Now().UTC() + + tests := []struct { + name string + record CommittedSnapshot + wantErr bool + }{ + { + name: "pending", + record: CommittedSnapshot{ + Snapshot: snapshot, + State: CommittedSnapshotStatePending, + }, + }, + { + name: "posted", + record: CommittedSnapshot{ + Snapshot: snapshot, + State: CommittedSnapshotStatePosted, + PostedAt: postedAt, + }, + }, + { + name: "unknown state", + record: CommittedSnapshot{ + Snapshot: snapshot, + }, + wantErr: true, + }, + { + name: "pending with post time", + record: CommittedSnapshot{ + Snapshot: snapshot, + State: CommittedSnapshotStatePending, + PostedAt: postedAt, + }, + wantErr: true, + }, + { + name: "posted without post time", + record: CommittedSnapshot{ + Snapshot: snapshot, + State: CommittedSnapshotStatePosted, + }, + wantErr: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + err := ValidateCommittedSnapshot(test.record) + if test.wantErr { + require.Error(t, err) + return + } + + require.NoError(t, err) + }) + } +} From 469b9dbfa4f2c3f9b4bbf08d9bfab6c9c3b0fb28 Mon Sep 17 00:00:00 2001 From: Joseph Chalabi Date: Mon, 27 Jul 2026 16:59:38 -0700 Subject: [PATCH 2/3] feat(gateway): serve committed inventory snapshots Expose only successfully posted snapshots by latest record or exact hash while keeping live challenge requests read-only. Signed-off-by: Joseph Chalabi --- cmd/provider-services/cmd/run.go | 14 ++- gateway/grpc/inventory.go | 58 +++++++++- gateway/grpc/inventory_test.go | 191 +++++++++++++++++++++++++++++++ gateway/grpc/server.go | 15 ++- 4 files changed, 273 insertions(+), 5 deletions(-) diff --git a/cmd/provider-services/cmd/run.go b/cmd/provider-services/cmd/run.go index eb8ea3296..94ae0e515 100644 --- a/cmd/provider-services/cmd/run.go +++ b/cmd/provider-services/cmd/run.go @@ -811,6 +811,11 @@ func doRunCmd(ctx context.Context, cmd *cobra.Command, _ []string) error { return err } + persistentConfig, err := fromctx.PersistentConfigFromCtx(ctx) + if err != nil { + return err + } + gwRest, err := gwrest.NewServer( ctx, logger, @@ -825,7 +830,14 @@ func doRunCmd(ctx context.Context, cmd *cobra.Command, _ []string) error { return err } - err = gwgrpc.NewServer(ctx, grpcaddr, accQuerier, service, snapshotter) + err = gwgrpc.NewServer( + ctx, + grpcaddr, + accQuerier, + service, + snapshotter, + persistentConfig.InventorySnapshots(), + ) if err != nil { return err } diff --git a/gateway/grpc/inventory.go b/gateway/grpc/inventory.go index 6d7129e41..62f5f5857 100644 --- a/gateway/grpc/inventory.go +++ b/gateway/grpc/inventory.go @@ -1,7 +1,9 @@ package grpc import ( + "bytes" "context" + "errors" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" @@ -17,7 +19,8 @@ type InventorySnapshotter interface { type grpcInventoryV1 struct { inventoryv1.UnimplementedInventoryServiceServer - snapshotter InventorySnapshotter + snapshotter InventorySnapshotter + committedSnapshots inventory.CommittedSnapshotReader } var _ inventoryv1.InventoryServiceServer = (*grpcInventoryV1)(nil) @@ -51,3 +54,56 @@ func (gm *grpcInventoryV1) GetInventorySnapshot(ctx context.Context, req *invent Provider: snapshot.Provider, }, nil } + +func (gm *grpcInventoryV1) GetCommittedInventorySnapshot(ctx context.Context, req *inventoryv1.GetCommittedInventorySnapshotRequest) (*inventoryv1.GetCommittedInventorySnapshotResponse, error) { + if req == nil { + return nil, status.Error(codes.InvalidArgument, "empty request") + } + + if gm.committedSnapshots == nil { + return nil, status.Error(codes.Unavailable, "committed inventory snapshot service unavailable") + } + + var ( + record inventory.CommittedSnapshot + err error + ) + + hash := req.GetSnapshotHash() + if len(hash) == 0 { + record, err = gm.committedSnapshots.Latest(ctx) + } else { + if err := inventory.ValidateCommittedSnapshotHash(hash); err != nil { + return nil, status.Error(codes.InvalidArgument, err.Error()) + } + + record, err = gm.committedSnapshots.Get(ctx, hash) + } + if err != nil { + switch { + case errors.Is(err, inventory.ErrCommittedSnapshotNotFound): + return nil, status.Error(codes.NotFound, err.Error()) + case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded): + return nil, status.FromContextError(err).Err() + default: + return nil, status.Error(codes.Internal, err.Error()) + } + } + if err := inventory.ValidateCommittedSnapshot(record); err != nil { + return nil, status.Error(codes.Internal, err.Error()) + } + if record.State != inventory.CommittedSnapshotStatePosted { + return nil, status.Error(codes.Internal, "committed inventory snapshot is not posted") + } + if len(hash) != 0 && !bytes.Equal(record.Snapshot.Hash, hash) { + return nil, status.Error(codes.Internal, "committed inventory snapshot hash does not match request") + } + + return &inventoryv1.GetCommittedInventorySnapshotResponse{ + SnapshotPayload: append([]byte(nil), record.Snapshot.Payload...), + Signature: append([]byte(nil), record.Snapshot.Signature...), + Provider: record.Snapshot.Provider, + SnapshotHash: append([]byte(nil), record.Snapshot.Hash...), + PostedAt: record.PostedAt, + }, nil +} diff --git a/gateway/grpc/inventory_test.go b/gateway/grpc/inventory_test.go index 0b4f9487d..54902f27e 100644 --- a/gateway/grpc/inventory_test.go +++ b/gateway/grpc/inventory_test.go @@ -5,6 +5,7 @@ import ( "context" "errors" "testing" + "time" "github.com/stretchr/testify/require" "google.golang.org/grpc/codes" @@ -142,3 +143,193 @@ func TestGetInventorySnapshotRejectsInvalidSnapshot(t *testing.T) { require.Equal(t, codes.Internal, status.Code(err)) require.Contains(t, err.Error(), "missing inventory snapshot payload") } + +type testCommittedSnapshotStore struct { + record inventory.CommittedSnapshot + err error + getHash []byte + getCalls int + latestCalls int + stageCalls int + markPostedCalls int + pendingCalls int +} + +func (s *testCommittedSnapshotStore) Stage(context.Context, inventory.Snapshot) error { + s.stageCalls++ + return nil +} + +func (s *testCommittedSnapshotStore) MarkPosted(context.Context, []byte, time.Time) error { + s.markPostedCalls++ + return nil +} + +func (s *testCommittedSnapshotStore) Pending(context.Context) ([]inventory.CommittedSnapshot, error) { + s.pendingCalls++ + return nil, nil +} + +func (s *testCommittedSnapshotStore) Get(_ context.Context, hash []byte) (inventory.CommittedSnapshot, error) { + s.getCalls++ + s.getHash = append([]byte(nil), hash...) + return inventory.CloneCommittedSnapshot(s.record), s.err +} + +func (s *testCommittedSnapshotStore) Latest(context.Context) (inventory.CommittedSnapshot, error) { + s.latestCalls++ + return inventory.CloneCommittedSnapshot(s.record), s.err +} + +func newPostedInventorySnapshot() inventory.CommittedSnapshot { + payload := []byte("committed payload") + + return inventory.CommittedSnapshot{ + Snapshot: inventory.Snapshot{ + Payload: payload, + Hash: inventory.HashPayload(payload), + Signature: []byte("committed signature"), + Provider: "akash1provider", + }, + State: inventory.CommittedSnapshotStatePosted, + PostedAt: time.Date(2026, time.July, 27, 19, 34, 56, 789, time.UTC), + } +} + +func TestGetInventorySnapshotNeverMutatesCommittedSnapshots(t *testing.T) { + for _, nonce := range [][]byte{nil, bytes.Repeat([]byte{1}, inventory.NonceSize)} { + store := &testCommittedSnapshotStore{} + snapshotter := &testInventorySnapshotter{ + snapshot: &inventory.Snapshot{ + Payload: []byte("payload"), + Hash: inventory.HashPayload([]byte("payload")), + Signature: []byte("signature"), + Provider: "akash1provider", + }, + } + server := &grpcInventoryV1{ + snapshotter: snapshotter, + committedSnapshots: store, + } + + resp, err := server.GetInventorySnapshot( + context.Background(), + &inventoryv1.GetInventorySnapshotRequest{Nonce: nonce}, + ) + require.NoError(t, err) + require.NotNil(t, resp) + require.Zero(t, store.getCalls) + require.Zero(t, store.latestCalls) + require.Zero(t, store.stageCalls) + require.Zero(t, store.markPostedCalls) + require.Zero(t, store.pendingCalls) + } +} + +func TestGetCommittedInventorySnapshotByHash(t *testing.T) { + record := newPostedInventorySnapshot() + store := &testCommittedSnapshotStore{record: record} + server := &grpcInventoryV1{committedSnapshots: store} + + resp, err := server.GetCommittedInventorySnapshot( + context.Background(), + &inventoryv1.GetCommittedInventorySnapshotRequest{SnapshotHash: record.Snapshot.Hash}, + ) + require.NoError(t, err) + require.Equal(t, record.Snapshot.Payload, resp.SnapshotPayload) + require.Equal(t, record.Snapshot.Signature, resp.Signature) + require.Equal(t, record.Snapshot.Provider, resp.Provider) + require.Equal(t, record.Snapshot.Hash, resp.SnapshotHash) + require.Equal(t, record.PostedAt, resp.PostedAt) + require.Equal(t, 1, store.getCalls) + require.Equal(t, record.Snapshot.Hash, store.getHash) + require.Zero(t, store.latestCalls) +} + +func TestGetCommittedInventorySnapshotLatest(t *testing.T) { + record := newPostedInventorySnapshot() + store := &testCommittedSnapshotStore{record: record} + server := &grpcInventoryV1{committedSnapshots: store} + + resp, err := server.GetCommittedInventorySnapshot( + context.Background(), + &inventoryv1.GetCommittedInventorySnapshotRequest{}, + ) + require.NoError(t, err) + require.Equal(t, record.Snapshot.Hash, resp.SnapshotHash) + require.Zero(t, store.getCalls) + require.Equal(t, 1, store.latestCalls) +} + +func TestGetCommittedInventorySnapshotRejectsInvalidRequest(t *testing.T) { + store := &testCommittedSnapshotStore{} + server := &grpcInventoryV1{committedSnapshots: store} + + resp, err := server.GetCommittedInventorySnapshot(context.Background(), nil) + require.Nil(t, resp) + require.Equal(t, codes.InvalidArgument, status.Code(err)) + + resp, err = server.GetCommittedInventorySnapshot( + context.Background(), + &inventoryv1.GetCommittedInventorySnapshotRequest{SnapshotHash: []byte("short")}, + ) + require.Nil(t, resp) + require.Equal(t, codes.InvalidArgument, status.Code(err)) + require.Zero(t, store.getCalls) + require.Zero(t, store.latestCalls) +} + +func TestGetCommittedInventorySnapshotReturnsUnavailableWithoutStore(t *testing.T) { + server := &grpcInventoryV1{} + + resp, err := server.GetCommittedInventorySnapshot( + context.Background(), + &inventoryv1.GetCommittedInventorySnapshotRequest{}, + ) + require.Nil(t, resp) + require.Equal(t, codes.Unavailable, status.Code(err)) +} + +func TestGetCommittedInventorySnapshotReturnsNotFound(t *testing.T) { + store := &testCommittedSnapshotStore{err: inventory.ErrCommittedSnapshotNotFound} + server := &grpcInventoryV1{committedSnapshots: store} + + resp, err := server.GetCommittedInventorySnapshot( + context.Background(), + &inventoryv1.GetCommittedInventorySnapshotRequest{}, + ) + require.Nil(t, resp) + require.Equal(t, codes.NotFound, status.Code(err)) +} + +func TestGetCommittedInventorySnapshotRejectsPendingRecord(t *testing.T) { + record := newPostedInventorySnapshot() + record.State = inventory.CommittedSnapshotStatePending + record.PostedAt = time.Time{} + server := &grpcInventoryV1{ + committedSnapshots: &testCommittedSnapshotStore{record: record}, + } + + resp, err := server.GetCommittedInventorySnapshot( + context.Background(), + &inventoryv1.GetCommittedInventorySnapshotRequest{}, + ) + require.Nil(t, resp) + require.Equal(t, codes.Internal, status.Code(err)) +} + +func TestGetCommittedInventorySnapshotRejectsMismatchedHash(t *testing.T) { + record := newPostedInventorySnapshot() + server := &grpcInventoryV1{ + committedSnapshots: &testCommittedSnapshotStore{record: record}, + } + + resp, err := server.GetCommittedInventorySnapshot( + context.Background(), + &inventoryv1.GetCommittedInventorySnapshotRequest{ + SnapshotHash: inventory.HashPayload([]byte("different payload")), + }, + ) + require.Nil(t, resp) + require.Equal(t, codes.Internal, status.Code(err)) +} diff --git a/gateway/grpc/server.go b/gateway/grpc/server.go index 3a140f3ba..2f4d5880d 100644 --- a/gateway/grpc/server.go +++ b/gateway/grpc/server.go @@ -27,6 +27,7 @@ import ( gwutils "github.com/akash-network/provider/gateway/utils" "github.com/akash-network/provider/tools/fromctx" ptypes "github.com/akash-network/provider/types" + "github.com/akash-network/provider/verification/inventory" ) type ContextKey string @@ -59,7 +60,14 @@ func ClaimsFromCtx(ctx context.Context) *ajwt.Claims { return val.(*ajwt.Claims) } -func NewServer(ctx context.Context, endpoint string, cquery gwutils.CertGetter, client provider.Client, snapshotter InventorySnapshotter) error { +func NewServer( + ctx context.Context, + endpoint string, + cquery gwutils.CertGetter, + client provider.Client, + snapshotter InventorySnapshotter, + committedSnapshots inventory.CommittedSnapshotReader, +) error { tlsCfg, err := gwutils.NewServerTLSConfig(ctx, cquery, endpoint) if err != nil { return err @@ -89,9 +97,10 @@ func NewServer(ctx context.Context, endpoint string, cquery gwutils.CertGetter, } leasev1.RegisterLeaseRPCServer(grpcSrv, leaseRPC) - if snapshotter != nil { + if snapshotter != nil || committedSnapshots != nil { inventoryv1.RegisterInventoryServiceServer(grpcSrv, &grpcInventoryV1{ - snapshotter: snapshotter, + snapshotter: snapshotter, + committedSnapshots: committedSnapshots, }) } gogoreflection.Register(grpcSrv) From 9f7897485043ca516fcf06fd61dd05b3ff1ecc8d Mon Sep 17 00:00:00 2001 From: Joseph Chalabi Date: Mon, 27 Jul 2026 19:52:43 -0700 Subject: [PATCH 3/3] feat(gateway): expose committed inventory status Expose metadata for the latest posted signed inventory snapshot through the existing provider status response. Missing or unreadable snapshot state leaves the legacy status response available. Signed-off-by: Joseph Chalabi --- cmd/provider-services/cmd/run.go | 5 ++ gateway/rest/router.go | 25 ++++-- gateway/rest/verification_inventory.go | 93 ++++++++++++++++++++ gateway/rest/verification_inventory_test.go | 96 +++++++++++++++++++++ 4 files changed, 212 insertions(+), 7 deletions(-) create mode 100644 gateway/rest/verification_inventory.go create mode 100644 gateway/rest/verification_inventory_test.go diff --git a/cmd/provider-services/cmd/run.go b/cmd/provider-services/cmd/run.go index 94ae0e515..825833ee9 100644 --- a/cmd/provider-services/cmd/run.go +++ b/cmd/provider-services/cmd/run.go @@ -816,6 +816,11 @@ func doRunCmd(ctx context.Context, cmd *cobra.Command, _ []string) error { return err } + gwrest.SetVerificationInventoryStatusSource( + clusterSettings, + persistentConfig.InventorySnapshots(), + ) + gwRest, err := gwrest.NewServer( ctx, logger, diff --git a/gateway/rest/router.go b/gateway/rest/router.go index ac386028d..49a7a2fbd 100644 --- a/gateway/rest/router.go +++ b/gateway/rest/router.go @@ -121,7 +121,7 @@ func newRouter(log log.Logger, addr sdk.Address, pclient provider.Client, ctxCon // GET /status // provider status endpoint does not require authentication router.HandleFunc("/status", - createStatusHandler(log, pclient, addr)). + createStatusHandler(log, pclient, addr, verificationInventoryStatusSourceFromConfig(ctxConfig))). Methods("GET") authedRouter := router.NewRoute().Subrouter() @@ -447,7 +447,12 @@ func createVersionHandler(log log.Logger, pclient provider.Client) http.HandlerF } } -func createStatusHandler(log log.Logger, sclient provider.StatusClient, providerAddr sdk.Address) http.HandlerFunc { +func createStatusHandler( + log log.Logger, + sclient provider.StatusClient, + providerAddr sdk.Address, + verificationInventory verificationInventoryStatusSource, +) http.HandlerFunc { return func(w http.ResponseWriter, req *http.Request) { status, err := sclient.Status(req.Context()) if err != nil { @@ -463,16 +468,22 @@ func createStatusHandler(log log.Logger, sclient provider.StatusClient, provider inventory := statusV1.Cluster.GetInventory() leasedIP = inventory.GetLeasedIP() } + verificationStatus, err := latestVerificationInventoryStatus(req.Context(), verificationInventory) + if err != nil { + log.Error("failed to fetch verification inventory status", "err", err) + } data := struct { // provider.Status apclient.ProviderStatus - Address string `json:"address"` - LeasedIP inventoryV1.ResourcePair `json:"leased_ip"` + Address string `json:"address"` + LeasedIP inventoryV1.ResourcePair `json:"leased_ip"` + VerificationInventory *verificationInventoryStatus `json:"verification_inventory,omitempty"` }{ - ProviderStatus: *status, - Address: providerAddr.String(), - LeasedIP: leasedIP, + ProviderStatus: *status, + Address: providerAddr.String(), + LeasedIP: leasedIP, + VerificationInventory: verificationStatus, } writeJSON(log, w, data) } diff --git a/gateway/rest/verification_inventory.go b/gateway/rest/verification_inventory.go new file mode 100644 index 000000000..f59dd68de --- /dev/null +++ b/gateway/rest/verification_inventory.go @@ -0,0 +1,93 @@ +package rest + +import ( + "context" + "encoding/base64" + "errors" + "time" + + inventoryv1 "pkg.akt.dev/go/inventory/v1" + + "github.com/akash-network/provider/verification/inventory" +) + +type verificationInventoryStatusSource interface { + Latest(context.Context) (inventory.CommittedSnapshot, error) +} + +type verificationInventoryStatus struct { + Provider string `json:"provider"` + Hash string `json:"hash"` + Signature string `json:"signature"` + SchemaVersion uint32 `json:"schema_version"` + CreatedAt time.Time `json:"created_at"` + PostedAt time.Time `json:"posted_at"` + Validation verificationInventoryValidation `json:"validation"` +} + +type verificationInventoryValidation struct { + Status string `json:"status"` + ValidatedAt time.Time `json:"validated_at"` +} + +type verificationInventoryStatusKey struct{} + +func SetVerificationInventoryStatusSource( + cfg map[interface{}]interface{}, + source verificationInventoryStatusSource, +) { + if cfg == nil || source == nil { + return + } + + cfg[verificationInventoryStatusKey{}] = source +} + +func verificationInventoryStatusSourceFromConfig( + cfg map[interface{}]interface{}, +) verificationInventoryStatusSource { + if cfg == nil { + return nil + } + + source, _ := cfg[verificationInventoryStatusKey{}].(verificationInventoryStatusSource) + return source +} + +func latestVerificationInventoryStatus( + ctx context.Context, + source verificationInventoryStatusSource, +) (*verificationInventoryStatus, error) { + if source == nil { + return nil, nil + } + + record, err := source.Latest(ctx) + if errors.Is(err, inventory.ErrCommittedSnapshotNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + if err := inventory.ValidateCommittedSnapshot(record); err != nil { + return nil, err + } + + var payload inventoryv1.SnapshotPayload + if err := payload.Unmarshal(record.Snapshot.Payload); err != nil { + return nil, err + } + + return &verificationInventoryStatus{ + Provider: record.Snapshot.Provider, + Hash: base64.StdEncoding.EncodeToString(record.Snapshot.Hash), + Signature: base64.StdEncoding.EncodeToString(record.Snapshot.Signature), + SchemaVersion: payload.SchemaVersion, + CreatedAt: payload.Timestamp, + PostedAt: record.PostedAt, + Validation: verificationInventoryValidation{ + Status: "valid", + ValidatedAt: record.PostedAt, + }, + }, nil +} diff --git a/gateway/rest/verification_inventory_test.go b/gateway/rest/verification_inventory_test.go new file mode 100644 index 000000000..9d2340518 --- /dev/null +++ b/gateway/rest/verification_inventory_test.go @@ -0,0 +1,96 @@ +package rest + +import ( + "context" + "encoding/base64" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + sdk "github.com/cosmos/cosmos-sdk/types" + + inventoryv1 "pkg.akt.dev/go/inventory/v1" + apclient "pkg.akt.dev/go/provider/client" + providerv1 "pkg.akt.dev/go/provider/v1" + "pkg.akt.dev/go/testutil" + + pmock "github.com/akash-network/provider/mocks/client" + "github.com/akash-network/provider/verification/inventory" +) + +type testVerificationInventoryStatusSource struct { + record inventory.CommittedSnapshot + err error +} + +func (s testVerificationInventoryStatusSource) Latest(context.Context) (inventory.CommittedSnapshot, error) { + return inventory.CloneCommittedSnapshot(s.record), s.err +} + +func TestStatusIncludesCommittedInventorySnapshot(t *testing.T) { + providerAddr := sdk.AccAddress(testutil.Key(t).PubKey().Address()) + createdAt := time.Date(2026, 7, 27, 12, 0, 0, 0, time.UTC) + postedAt := createdAt.Add(time.Minute) + payload, err := (&inventoryv1.SnapshotPayload{ + SchemaVersion: 1, + Provider: providerAddr.String(), + ChainID: "sandbox-2", + Timestamp: createdAt, + }).Marshal() + require.NoError(t, err) + + snapshot := inventory.Snapshot{ + Payload: payload, + Hash: inventory.HashPayload(payload), + Signature: []byte("provider-signature"), + Provider: providerAddr.String(), + } + record, err := inventory.NewPendingCommittedSnapshot(snapshot) + require.NoError(t, err) + record, err = inventory.MarkCommittedSnapshotPosted(record, postedAt) + require.NoError(t, err) + + pclient := pmock.NewClient(t) + pclient.On("Status", mock.Anything).Return(&apclient.ProviderStatus{}, nil) + pclient.On("StatusV1", mock.Anything).Return(&providerv1.Status{}, nil) + + req := httptest.NewRequest(http.MethodGet, "/status", nil) + resp := httptest.NewRecorder() + createStatusHandler( + testutil.Logger(t), + pclient, + providerAddr, + testVerificationInventoryStatusSource{record: record}, + ).ServeHTTP(resp, req) + + require.Equal(t, http.StatusOK, resp.Code) + + var body struct { + VerificationInventory *verificationInventoryStatus `json:"verification_inventory"` + } + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &body)) + require.NotNil(t, body.VerificationInventory) + require.Equal(t, providerAddr.String(), body.VerificationInventory.Provider) + require.Equal(t, base64.StdEncoding.EncodeToString(snapshot.Hash), body.VerificationInventory.Hash) + require.Equal(t, base64.StdEncoding.EncodeToString(snapshot.Signature), body.VerificationInventory.Signature) + require.Equal(t, uint32(1), body.VerificationInventory.SchemaVersion) + require.Equal(t, createdAt, body.VerificationInventory.CreatedAt) + require.Equal(t, postedAt, body.VerificationInventory.PostedAt) + require.Equal(t, "valid", body.VerificationInventory.Validation.Status) + require.Equal(t, postedAt, body.VerificationInventory.Validation.ValidatedAt) +} + +func TestLatestVerificationInventoryStatusIgnoresMissingSnapshot(t *testing.T) { + status, err := latestVerificationInventoryStatus( + context.Background(), + testVerificationInventoryStatusSource{err: inventory.ErrCommittedSnapshotNotFound}, + ) + + require.NoError(t, err) + require.Nil(t, status) +}