feat(inventory): persist committed snapshots

cmd/provider-services/cmd/run.go

@@ -811,6 +811,16 @@ func doRunCmd(ctx context.Context, cmd *cobra.Command, _ []string) error {
 		return err
 	}
 
+	persistentConfig, err := fromctx.PersistentConfigFromCtx(ctx)
+	if err != nil {
+		return err
+	}
+
+	gwrest.SetVerificationInventoryStatusSource(
+		clusterSettings,
+		persistentConfig.InventorySnapshots(),
+	)
+
 	gwRest, err := gwrest.NewServer(
 		ctx,
 		logger,
@@ -825,7 +835,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
 	}

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
+}

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))
+}

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)

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)
 	}