Cosmos x/vstream implementation

golang/cosmos/proto/agoric/types/stream.proto

@@ -0,0 +1,74 @@
+syntax = "proto3";
+package agoric.types;
+
+import "gogoproto/gogo.proto";
+
+option go_package = "github.com/Agoric/agoric-sdk/golang/cosmos/types/stream";
+
+// StreamCell represents one publish state in a vstream stream.
+message StreamCell {
+    option (gogoproto.equal) = false;
+
+    // EndState is the state of the stream at the last position in this cell.
+    enum EndState {
+      // END_STATE_UNSPECIFIED means unknown end state.
+      END_STATE_UNSPECIFIED = 0;
+      // END_STATE_APPENDABLE means that appending to this cell is possible.
+      END_STATE_APPENDABLE = 1;
+      // END_STATE_FINISHED means this cell has terminated.
+      END_STATE_FINISHED = 2;
+      // END_STATE_FAILURE means this cell has terminated with an error value.
+      END_STATE_FAILURE = 3;
+    }
+
+    // Chronologically-ordered list of values for this cell.
+    // If state=END_STATE_FAILURE, the last value is the serialised error.
+    repeated bytes values = 1;
+
+    // The block height in which this cell was last updated.
+    int64 updated_block_height = 2 [
+        (gogoproto.jsontag)    = "updated_block_height",
+        (gogoproto.moretags)   = "yaml:\"updated_block_height\""
+    ];
+
+    // The state as of the last value.
+    EndState end_state = 3 [
+        (gogoproto.jsontag)    = "end_state",
+        (gogoproto.moretags)   = "yaml:\"end_state\""
+    ];
+
+    // The prior position in the stream.  If this is the first cell in the
+    // stream, this will be the default value. All other cells will have a
+    // reference with a nonzero block height.
+    StreamPosition prior = 4 [
+      (gogoproto.nullable) = false
+    ];
+}
+
+// StreamPosition is data that refers to a particular stream cell value.
+message StreamPosition {
+    // The height at which the cell is stored.
+    int64 block_height = 1 [
+        (gogoproto.jsontag)    = "block_height",
+        (gogoproto.moretags)   = "yaml:\"block_height\""
+    ];
+
+    // The low-level (Cosmos SDK multistore) store name under which the cell is stored.
+    string store_name = 2 [
+        (gogoproto.jsontag)    = "store_name",
+        (gogoproto.moretags)   = "yaml:\"store_name\""
+    ];
+
+    // The low-level (Cosmos SDK multistore) store subkey under which the cell is stored.
+    bytes store_subkey = 3 [
+        (gogoproto.jsontag)    = "store_subkey",
+        (gogoproto.moretags)   = "yaml:\"store_subkey\""
+    ];
+
+    // The sequence number of the value at this position.  Necessary to allow
+    // references to a fork point within a cell.
+    uint64 sequence_number = 4 [
+        (gogoproto.jsontag)    = "sequence_number",
+        (gogoproto.moretags)   = "yaml:\"sequence_number\""
+    ];
+}

golang/cosmos/types/state_ref.go

@@ -0,0 +1,57 @@
+package types
+
+import (
+	"fmt"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+)
+
+// An interface to allow updating KVStore-backed state and extracting proof
+// parameters.
+type StateRef interface {
+	Read(ctx sdk.Context) ([]byte, error)
+	Write(ctx sdk.Context, value []byte) error
+	Exists(ctx sdk.Context) bool
+	StoreName() string
+	StoreSubkey() []byte
+	String() string
+}
+
+type KVStoreStateRef struct {
+	storeKey sdk.StoreKey
+	subkey   []byte
+}
+
+var _ StateRef = KVStoreStateRef{}
+
+func NewKVStoreStateRef(storeKey sdk.StoreKey, subkey []byte) *KVStoreStateRef {
+	return &KVStoreStateRef{storeKey, subkey}
+}
+
+func (s KVStoreStateRef) Read(ctx sdk.Context) ([]byte, error) {
+	store := ctx.KVStore(s.storeKey)
+	return store.Get(s.subkey), nil
+}
+
+func (s KVStoreStateRef) Write(ctx sdk.Context, value []byte) error {
+	store := ctx.KVStore(s.storeKey)
+	store.Set(s.subkey, value)
+	return nil
+}
+
+func (s KVStoreStateRef) Exists(ctx sdk.Context) bool {
+	store := ctx.KVStore(s.storeKey)
+	return store.Has(s.subkey)
+}
+
+func (s KVStoreStateRef) StoreName() string {
+	return s.storeKey.Name()
+}
+
+func (s KVStoreStateRef) StoreSubkey() []byte {
+	return s.subkey
+}
+
+func (s KVStoreStateRef) String() string {
+	return fmt.Sprintf("KVStoreStateRef{%s, %s}", s.storeKey.Name(), s.subkey)
+}

golang/cosmos/types/stream/cell.go

@@ -0,0 +1,59 @@
+package stream
+
+import (
+	"encoding/json"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+
+	agoric "github.com/Agoric/agoric-sdk/golang/cosmos/types"
+)
+
+// NewStreamCell creates a new StreamCell at blockHeight with the specified
+// prior position.
+func NewStreamCell(blockHeight int64, prior StreamPosition) StreamCell {
+	return StreamCell{
+		UpdatedBlockHeight: blockHeight,
+		EndState:           StreamCell_END_STATE_APPENDABLE,
+		Values:             make([][]byte, 0, 1),
+		Prior:              prior,
+	}
+}
+
+func NewStreamPosition(blockHeight int64, storeName string, subkey []byte, seq uint64) StreamPosition {
+	return StreamPosition{
+		BlockHeight:    blockHeight,
+		StoreName:      storeName,
+		StoreSubkey:    subkey,
+		SequenceNumber: seq,
+	}
+}
+
+func NewZeroStreamPosition() StreamPosition {
+	return NewStreamPosition(0, "", nil, 0)
+}
+
+// GetLatestPosition returns the position of the last value in the cell, or a
+// zero stream position if the referenced cell does not exist.
+func GetLatestPosition(ctx sdk.Context, state agoric.StateRef) (*StreamPosition, error) {
+	if !state.Exists(ctx) {
+		zeroPosition := NewZeroStreamPosition()
+		return &zeroPosition, nil
+	}
+
+	var priorCell StreamCell
+	data, err := state.Read(ctx)
+	if err != nil {
+		return nil, err
+	}
+	if err := json.Unmarshal(data, &priorCell); err != nil {
+		return nil, err
+	}
+
+	priorReference := NewStreamPosition(
+		priorCell.UpdatedBlockHeight,
+		state.StoreName(),
+		state.StoreSubkey(),
+		uint64(len(priorCell.Values)),
+	)
+	return &priorReference, nil
+}

golang/cosmos/types/stream/operation.go

@@ -0,0 +1,188 @@
+package stream
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+
+	agoric "github.com/Agoric/agoric-sdk/golang/cosmos/types"
+)
+
+type StreamCellUpdater interface {
+	Update(cell StreamCell) error
+}
+
+var _, _ StreamCellUpdater = AppendStreamCellUpdater{}, FailStreamCellUpdater{}
+
+type StreamOperation struct {
+	state   agoric.StateRef
+	Updater StreamCellUpdater
+	Prior   StreamPosition
+}
+
+func NewStreamOperation(ctx sdk.Context, state agoric.StateRef, updater StreamCellUpdater) StreamOperation {
+	return StreamOperation{
+		state:   state,
+		Updater: updater,
+		Prior:   NewZeroStreamPosition(),
+	}
+}
+
+func (so StreamOperation) GetLatestPosition(ctx sdk.Context) (*StreamPosition, error) {
+	return GetLatestPosition(ctx, so.state)
+}
+
+// LoadAndCheckHead returns a head satisfying prior that we can mutate.
+func (so StreamOperation) LoadAndCheckHead(ctx sdk.Context, prior StreamPosition) (*StreamCell, error) {
+	head := NewStreamCell(ctx.BlockHeight(), prior)
+	if !so.state.Exists(ctx) {
+		// No prior state, safe to use the fresh head.
+		return &head, nil
+	}
+	// Get the current head.
+	data, err := so.state.Read(ctx)
+	if err != nil {
+		return nil, err
+	}
+	if err := json.Unmarshal([]byte(data), &head); err != nil {
+		return nil, err
+	}
+	if prior.BlockHeight != head.UpdatedBlockHeight {
+		return nil, fmt.Errorf("prior block height %d does not match current %q head block height %d", prior.BlockHeight, so.state, head.UpdatedBlockHeight)
+	}
+	nextSequence := head.Prior.SequenceNumber + uint64(len(head.Values))
+	if prior.SequenceNumber != nextSequence {
+		return nil, fmt.Errorf("prior sequence number %d does not point to last value %d", prior.SequenceNumber, nextSequence)
+	}
+	stateStoreName := so.state.StoreName()
+	if prior.StoreName != stateStoreName {
+		return nil, fmt.Errorf("prior store name %s does not match state store name %s", prior.StoreName, stateStoreName)
+	}
+	stateStoreSubKey := so.state.StoreSubkey()
+	if !bytes.Equal(prior.StoreSubkey, stateStoreSubKey) {
+		return nil, fmt.Errorf("prior store subkey %s does not match state store subkey %s", prior.StoreSubkey, stateStoreSubKey)
+	}
+	// We can update the current head state.
+	return &head, nil
+}
+
+// Commit commits the stream operation to the current state.
+func (so StreamOperation) Commit(ctx sdk.Context, priorPos StreamPosition, forceOverwrite bool) error {
+	head := NewStreamCell(ctx.BlockHeight(), priorPos)
+	if !forceOverwrite {
+		// Get the current head and assert that it's compatible with prior.
+		headP, err := so.LoadAndCheckHead(ctx, priorPos)
+		if err != nil {
+			return err
+		}
+		head = *headP
+	}
+
+	if head.UpdatedBlockHeight != ctx.BlockHeight() {
+		// Start a new values list.
+		head.UpdatedBlockHeight = ctx.BlockHeight()
+		head.Values = nil
+	}
+
+	head.Prior = priorPos
+
+	if head.Values == nil {
+		// Allocate fresh values for potential appending.
+		head.Values = make([][]byte, 0, 1)
+	}
+
+	switch head.EndState {
+	case StreamCell_END_STATE_APPENDABLE:
+		break
+	case StreamCell_END_STATE_FINISHED:
+		return fmt.Errorf("cannot update stream at %s that is already done", so.state)
+	case StreamCell_END_STATE_FAILURE:
+		return fmt.Errorf("cannot update stream at %s that has an error: %s", so.state, head.Values[len(head.Values)-1])
+	default:
+		return fmt.Errorf("cannot update stream at %s with unrecognized end state %q", so.state, head.EndState)
+	}
+
+	if err := so.Updater.Update(head); err != nil {
+		return err
+	}
+
+	// Convert the head to JSON.
+	bz, err := json.Marshal(head)
+	if err != nil {
+		return err
+	}
+
+	// COMMIT POINT
+	// Store the marshalled stream cell.
+	if err := so.state.Write(ctx, bz); err != nil {
+		return err
+	}
+
+	// Emit the advisory state change event.
+	ctx.EventManager().EmitEvent(
+		agoric.NewStateChangeEvent(
+			so.state.StoreName(),
+			so.state.StoreSubkey(),
+			bz,
+		),
+	)
+	return nil
+}
+
+// CommitToCurrent fetches the latest position and appends to it.
+func (so StreamOperation) CommitToCurrent(ctx sdk.Context) error {
+	prior, err := so.GetLatestPosition(ctx)
+	if err != nil {
+		return err
+	}
+	return so.Commit(ctx, *prior, false)
+}
+
+type AppendStreamCellUpdater struct {
+	Done  bool
+	Value []byte
+}
+
+func NewUpdateStreamOperation(ctx sdk.Context, state agoric.StateRef, value []byte) StreamOperation {
+	updater := AppendStreamCellUpdater{
+		Value: value,
+		Done:  false,
+	}
+	return NewStreamOperation(ctx, state, updater)
+}
+
+func NewFinishStreamOperation(ctx sdk.Context, state agoric.StateRef, value []byte) StreamOperation {
+	updater := AppendStreamCellUpdater{
+		Value: value,
+		Done:  true,
+	}
+	return NewStreamOperation(ctx, state, updater)
+}
+
+func (op AppendStreamCellUpdater) Update(cell StreamCell) error {
+	// Add the new state to the batch.
+	cell.Values = append(cell.Values, op.Value)
+	if op.Done {
+		cell.EndState = StreamCell_END_STATE_FINISHED
+	}
+	return nil
+}
+
+type FailStreamCellUpdater struct {
+	Failure []byte
+}
+
+func (op FailStreamCellUpdater) Update(cell StreamCell) error {
+	cell.Values = append(cell.Values, op.Failure)
+	cell.EndState = StreamCell_END_STATE_FAILURE
+	return nil
+}
+
+func NewFailStreamOperation(ctx sdk.Context, state agoric.StateRef, failure []byte) StreamOperation {
+	updater := FailStreamCellUpdater{
+		Failure: failure,
+	}
+	return NewStreamOperation(ctx, state, updater)
+}