go/consensus: use typed attributes in all services

.changelog/4034.breaking.md

@@ -0,0 +1,4 @@
+Use typed attributes in all consensus services
+
+Beacon, keymanager, registry, roothash and scheduler events are updated to use
+the typed event attribute API.

go/beacon/api/api.go

@@ -186,3 +186,25 @@ func (g *Genesis) SanityCheck() error {
 
 	return nil
 }
+
+// EpochEvent is the epoch event.
+type EpochEvent struct {
+	// Epoch is the new epoch.
+	Epoch EpochTime `json:"epoch,omitempty"`
+}
+
+// EventKind returns a string representation of this event's kind.
+func (ev *EpochEvent) EventKind() string {
+	return "epoch"
+}
+
+// BeaconEvent is the beacon event.
+type BeaconEvent struct {
+	// Beacon is the new beacon value.
+	Beacon []byte `json:"beacon,omitempty"`
+}
+
+// EventKind returns a string representation of this event's kind.
+func (ev *BeaconEvent) EventKind() string {
+	return "beacon"
+}

go/consensus/tendermint/api/api.go

@@ -85,12 +85,21 @@ func NodeToP2PAddr(n *node.Node) (*tmp2p.NetAddress, error) {
 }
 
 // TypedAttribute is an interface implemented by types which can be transparently used as event
-// attributes.
+// attributes with CBOR-marshalled value.
 type TypedAttribute interface {
 	// EventKind returns a string representation of this event's kind.
 	EventKind() string
 }
 
+// CustomTypedAttribute is an interface implemented by types which can be transparently used as event
+// attributes with custom value encoding.
+type CustomTypedAttribute interface {
+	TypedAttribute
+
+	// EventValue returns a byte representation of this events value.
+	EventValue() []byte
+}
+
 // IsAttributeKind checks whether the given attribute key corresponds to the passed typed attribute.
 func IsAttributeKind(key []byte, kind TypedAttribute) bool {
 	return bytes.Equal(key, []byte(kind.EventKind()))
@@ -102,8 +111,8 @@ type EventBuilder struct {
 	ev  types.Event
 }
 
-// Attribute appends a key/value pair to the event.
-func (bld *EventBuilder) Attribute(key, value []byte) *EventBuilder {
+// attribute appends a key/value pair to the event.
+func (bld *EventBuilder) attribute(key, value []byte) *EventBuilder {
 	bld.ev.Attributes = append(bld.ev.Attributes, types.EventAttribute{
 		Key:   key,
 		Value: value,
@@ -117,7 +126,12 @@ func (bld *EventBuilder) Attribute(key, value []byte) *EventBuilder {
 // The typed attribute is automatically converted to a key/value pair where its EventKind is used
 // as the key and a CBOR-marshalled value is used as value.
 func (bld *EventBuilder) TypedAttribute(value TypedAttribute) *EventBuilder {
-	return bld.Attribute([]byte(value.EventKind()), cbor.Marshal(value))
+	return bld.attribute([]byte(value.EventKind()), cbor.Marshal(value))
+}
+
+// CustomTypedAttribute appends a typed attribute to the event.
+func (bld *EventBuilder) CustomTypedAttribute(value CustomTypedAttribute) *EventBuilder {
+	return bld.attribute([]byte(value.EventKind()), value.EventValue())
 }
 
 // Dirty returns true iff the EventBuilder has attributes.

go/consensus/tendermint/api/context.go

@@ -329,8 +329,8 @@ func (c *Context) GetEvents() []types.Event {
 	return c.events
 }
 
-// HasEvent checks if a specific event has been emitted.
-func (c *Context) HasEvent(app string, key []byte) bool {
+// hasEvent checks if a specific event has been emitted.
+func (c *Context) hasEvent(app string, key []byte) bool {
 	evType := EventTypeForApp(app)
 
 	for _, ev := range c.events {
@@ -349,7 +349,7 @@ func (c *Context) HasEvent(app string, key []byte) bool {
 
 // HasTypedEvent checks if a specific typed event has been emitted.
 func (c *Context) HasTypedEvent(app string, kind TypedAttribute) bool {
-	return c.HasEvent(app, []byte(kind.EventKind()))
+	return c.hasEvent(app, []byte(kind.EventKind()))
 }
 
 // DecodeTypedEvent decodes the given raw event as a specific typed event.

go/consensus/tendermint/api/context_test.go

@@ -16,6 +16,17 @@ func (k testBlockContextKey) NewDefault() interface{} {
 	return 42
 }
 
+// FooEvent is a test event.
+type FooEvent struct {
+	// Bar is the test event value.
+	Bar []byte
+}
+
+// EventKind returns a string representation of this event's kind.
+func (ev *FooEvent) EventKind() string {
+	return "foo"
+}
+
 func TestBlockContext(t *testing.T) {
 	require := require.New(t)
 
@@ -57,7 +68,7 @@ func TestChildContext(t *testing.T) {
 	require.EqualValues(ctx.BlockContext(), child.BlockContext(), "child.BlockContext should correspond to parent.BlockContext")
 
 	// Emitting an event should not propagate to the parent immediately.
-	child.EmitEvent(NewEventBuilder("test").Attribute([]byte("foo"), []byte("bar")))
+	child.EmitEvent(NewEventBuilder("test").TypedAttribute(&FooEvent{Bar: []byte("bar")}))
 	require.Len(child.GetEvents(), 1, "child event should be stored")
 	require.Len(ctx.GetEvents(), 0, "child event should not immediately propagate")
 	events := child.GetEvents()
@@ -70,7 +81,7 @@ func TestChildContext(t *testing.T) {
 	defer ctx.Close()
 
 	child = ctx.WithSimulation()
-	child.EmitEvent(NewEventBuilder("test").Attribute([]byte("foo"), []byte("bar")))
+	child.EmitEvent(NewEventBuilder("test").TypedAttribute(&FooEvent{Bar: []byte("bar")}))
 	child.Close()
 	require.Empty(ctx.GetEvents(), "events should not propagate in simulation mode")
 
@@ -91,7 +102,7 @@ func TestTransactionContext(t *testing.T) {
 	child := ctx.NewTransaction()
 
 	// Emitted events and state updates should not propagate to the parent unless committed.
-	child.EmitEvent(NewEventBuilder("test").Attribute([]byte("foo"), []byte("bar")))
+	child.EmitEvent(NewEventBuilder("test").TypedAttribute(&FooEvent{Bar: []byte("bar")}))
 	require.Len(child.GetEvents(), 1, "child event should be stored")
 	require.Len(ctx.GetEvents(), 0, "child event should not immediately propagate")
 
@@ -113,7 +124,7 @@ func TestTransactionContext(t *testing.T) {
 
 	child = ctx.NewTransaction()
 
-	child.EmitEvent(NewEventBuilder("test").Attribute([]byte("foo"), []byte("bar")))
+	child.EmitEvent(NewEventBuilder("test").TypedAttribute(&FooEvent{Bar: []byte("bar")}))
 	require.Len(child.GetEvents(), 1, "child event should be stored")
 	require.Len(ctx.GetEvents(), 0, "child event should not immediately propagate")
 	events := child.GetEvents()

go/consensus/tendermint/apps/beacon/api.go

@@ -27,12 +27,6 @@ var (
 	// beacon application.
 	QueryApp = api.QueryForApp(AppName)
 
-	// KeyEpoch is the ABCI event attribute for specifying the set epoch.
-	KeyEpoch = []byte("epoch")
-
-	// KeyBeacon is the ABCI event attribute key for the new beacons.
-	KeyBeacon = []byte("beacon")
-
 	// MethodSetEpoch is the method name for setting epochs.
 	MethodSetEpoch = transaction.NewMethodName(AppName, "SetEpoch", beacon.EpochTime(0))
 

go/consensus/tendermint/apps/beacon/beacon.go

@@ -10,7 +10,6 @@ import (
 	"golang.org/x/crypto/sha3"
 
 	beacon "github.com/oasisprotocol/oasis-core/go/beacon/api"
-	"github.com/oasisprotocol/oasis-core/go/common/cbor"
 	"github.com/oasisprotocol/oasis-core/go/consensus/api/transaction"
 	"github.com/oasisprotocol/oasis-core/go/consensus/tendermint/api"
 	beaconState "github.com/oasisprotocol/oasis-core/go/consensus/tendermint/apps/beacon/state"
@@ -90,7 +89,7 @@ func (app *beaconApplication) EndBlock(ctx *api.Context, req types.RequestEndBlo
 }
 
 func (app *beaconApplication) doEmitEpochEvent(ctx *api.Context, epoch beacon.EpochTime) {
-	ctx.EmitEvent(api.NewEventBuilder(app.Name()).Attribute(KeyEpoch, cbor.Marshal(epoch)))
+	ctx.EmitEvent(api.NewEventBuilder(app.Name()).TypedAttribute(&beacon.EpochEvent{Epoch: epoch}))
 }
 
 func (app *beaconApplication) scheduleEpochTransitionBlock(
@@ -111,17 +110,17 @@ func (app *beaconApplication) scheduleEpochTransitionBlock(
 	return nil
 }
 
-func (app *beaconApplication) onNewBeacon(ctx *api.Context, beacon []byte) error {
+func (app *beaconApplication) onNewBeacon(ctx *api.Context, value []byte) error {
 	state := beaconState.NewMutableState(ctx.State())
 
-	if err := state.SetBeacon(ctx, beacon); err != nil {
+	if err := state.SetBeacon(ctx, value); err != nil {
 		ctx.Logger().Error("onNewBeacon: failed to set beacon",
 			"err", err,
 		)
 		return fmt.Errorf("beacon: failed to set beacon: %w", err)
 	}
 
-	ctx.EmitEvent(api.NewEventBuilder(app.Name()).Attribute(KeyBeacon, beacon))
+	ctx.EmitEvent(api.NewEventBuilder(app.Name()).TypedAttribute(&beacon.BeaconEvent{Beacon: value}))
 
 	return nil
 }

go/consensus/tendermint/apps/keymanager/api.go

@@ -18,8 +18,4 @@ var (
 	// QueryApp is a query for filtering transactions processed by the
 	// key manager application.
 	QueryApp = api.QueryForApp(AppName)
-
-	// KeyStatusUpdate is an ABCI event attribute key for a key manager
-	// status update (value is a CBOR serialized key manager status).
-	KeyStatusUpdate = []byte("status")
 )

go/consensus/tendermint/apps/keymanager/genesis.go

@@ -9,7 +9,6 @@ import (
 	"github.com/tendermint/tendermint/abci/types"
 
 	"github.com/oasisprotocol/oasis-core/go/common"
-	"github.com/oasisprotocol/oasis-core/go/common/cbor"
 	tmapi "github.com/oasisprotocol/oasis-core/go/consensus/tendermint/api"
 	keymanagerState "github.com/oasisprotocol/oasis-core/go/consensus/tendermint/apps/keymanager/state"
 	genesis "github.com/oasisprotocol/oasis-core/go/genesis/api"
@@ -78,7 +77,9 @@ func (app *keymanagerApplication) InitChain(ctx *tmapi.Context, request types.Re
 	}
 
 	if len(toEmit) > 0 {
-		ctx.EmitEvent(tmapi.NewEventBuilder(app.Name()).Attribute(KeyStatusUpdate, cbor.Marshal(toEmit)))
+		ctx.EmitEvent(tmapi.NewEventBuilder(app.Name()).TypedAttribute(&keymanager.StatusUpdateEvent{
+			Statuses: toEmit,
+		}))
 	}
 
 	return nil

go/consensus/tendermint/apps/keymanager/keymanager.go

@@ -185,7 +185,9 @@ func (app *keymanagerApplication) onEpochChange(ctx *tmapi.Context, epoch beacon
 
 	// Emit the update event if required.
 	if len(toEmit) > 0 {
-		ctx.EmitEvent(tmapi.NewEventBuilder(app.Name()).Attribute(KeyStatusUpdate, cbor.Marshal(toEmit)))
+		ctx.EmitEvent(tmapi.NewEventBuilder(app.Name()).TypedAttribute(&api.StatusUpdateEvent{
+			Statuses: toEmit,
+		}))
 	}
 
 	return nil

go/consensus/tendermint/apps/keymanager/transactions.go

@@ -3,7 +3,6 @@ package keymanager
 import (
 	"fmt"
 
-	"github.com/oasisprotocol/oasis-core/go/common/cbor"
 	tmapi "github.com/oasisprotocol/oasis-core/go/consensus/tendermint/api"
 	keymanagerState "github.com/oasisprotocol/oasis-core/go/consensus/tendermint/apps/keymanager/state"
 	registryState "github.com/oasisprotocol/oasis-core/go/consensus/tendermint/apps/registry/state"
@@ -81,7 +80,9 @@ func (app *keymanagerApplication) updatePolicy(
 		panic(fmt.Errorf("failed to set keymanager status: %w", err))
 	}
 
-	ctx.EmitEvent(tmapi.NewEventBuilder(app.Name()).Attribute(KeyStatusUpdate, cbor.Marshal([]*api.Status{newStatus})))
+	ctx.EmitEvent(tmapi.NewEventBuilder(app.Name()).TypedAttribute(&api.StatusUpdateEvent{
+		Statuses: []*api.Status{newStatus},
+	}))
 
 	return nil
 }

go/consensus/tendermint/apps/registry/api.go

@@ -1,7 +1,6 @@
 package registry
 
 import (
-	"github.com/oasisprotocol/oasis-core/go/common/entity"
 	"github.com/oasisprotocol/oasis-core/go/consensus/tendermint/api"
 )
 
@@ -20,40 +19,4 @@ var (
 	// QueryApp is a query for filtering events processed by
 	// the registry application.
 	QueryApp = api.QueryForApp(AppName)
-
-	// KeyRuntimeRegistered is the ABCI event attribute for new
-	// runtime registrations (value is the CBOR serialized runtime
-	// descriptor).
-	KeyRuntimeRegistered = []byte("runtime.registered")
-
-	// KeyEntityRegistered is the ABCI event attribute for new entity
-	// registrations (value is the CBOR serialized entity descriptor).
-	KeyEntityRegistered = []byte("entity.registered")
-
-	// KeyEntityDeregistered is the ABCI event attribute for entity
-	// deregistrations (value is a CBOR serialized EntityDeregistration).
-	KeyEntityDeregistered = []byte("entity.deregistered")
-
-	// KeyNodeRegistered is the ABCI event attribute for new node
-	// registrations (value is the CBOR serialized node descriptor).
-	KeyNodeRegistered = []byte("nodes.registered")
-
-	// KeyNodesExpired is the ABCI event attribute for node
-	// deregistrations due to expiration (value is a CBOR serialized
-	// vector of node descriptors).
-	KeyNodesExpired = []byte("nodes.expired")
-
-	// KeyNodeUnfrozen is the ABCI event attribute for when nodes
-	// become unfrozen (value is CBOR serialized node ID).
-	KeyNodeUnfrozen = []byte("nodes.unfrozen")
-
-	// KeyRegistryNodeListEpoch is the ABCI event attribute for
-	// registry epochs.
-	KeyRegistryNodeListEpoch = []byte("nodes.epoch")
 )
-
-// EntityDeregistration is an entity deregistration.
-type EntityDeregistration struct {
-	// Deregistered entity.
-	Entity entity.Entity `json:"entity"`
-}

go/consensus/tendermint/apps/registry/registry.go

@@ -222,18 +222,12 @@ func (app *registryApplication) onRegistryEpochChanged(ctx *api.Context, registr
 		}
 	}
 
-	// Emit the RegistryNodeListEpoch notification event.
-	evb := api.NewEventBuilder(app.Name())
-	// (Dummy value, should be ignored.)
-	evb = evb.Attribute(KeyRegistryNodeListEpoch, []byte("1"))
-
-	if len(expiredNodes) > 0 {
-		// Iff any nodes have expired, force-emit the NodesExpired event
-		// so the change is picked up.
-		evb = evb.Attribute(KeyNodesExpired, cbor.Marshal(expiredNodes))
+	// Emit the expired node event for all expired nodes.
+	for _, expiredNode := range expiredNodes {
+		ctx.EmitEvent(api.NewEventBuilder(app.Name()).TypedAttribute(&registry.NodeEvent{Node: expiredNode, IsRegistration: false}))
 	}
-
-	ctx.EmitEvent(evb)
+	// Emit the node list epoch event.
+	ctx.EmitEvent(api.NewEventBuilder(app.Name()).TypedAttribute(&registry.NodeListEpochEvent{}))
 
 	return nil
 }

go/consensus/tendermint/apps/registry/transactions.go

@@ -77,7 +77,7 @@ func (app *registryApplication) registerEntity(
 		"entity", ent,
 	)
 
-	ctx.EmitEvent(api.NewEventBuilder(app.Name()).Attribute(KeyEntityRegistered, cbor.Marshal(ent)))
+	ctx.EmitEvent(api.NewEventBuilder(app.Name()).TypedAttribute(&registry.EntityEvent{Entity: ent, IsRegistration: true}))
 
 	return nil
 }
@@ -150,10 +150,7 @@ func (app *registryApplication) deregisterEntity(ctx *api.Context, state *regist
 		"entity_id", id,
 	)
 
-	tagV := &EntityDeregistration{
-		Entity: *removedEntity,
-	}
-	ctx.EmitEvent(api.NewEventBuilder(app.Name()).Attribute(KeyEntityDeregistered, cbor.Marshal(tagV)))
+	ctx.EmitEvent(api.NewEventBuilder(app.Name()).TypedAttribute(&registry.EntityEvent{Entity: removedEntity, IsRegistration: false}))
 
 	return nil
 }
@@ -514,7 +511,7 @@ func (app *registryApplication) registerNode( // nolint: gocyclo
 				return err
 			}
 
-			ctx.EmitEvent(api.NewEventBuilder(app.Name()).Attribute(KeyRuntimeRegistered, cbor.Marshal(rt)))
+			ctx.EmitEvent(api.NewEventBuilder(app.Name()).TypedAttribute(&registry.RuntimeEvent{Runtime: rt}))
 		case registry.ErrNoSuchRuntime:
 			// Runtime was not suspended.
 		default:
@@ -531,7 +528,7 @@ func (app *registryApplication) registerNode( // nolint: gocyclo
 		"roles", newNode.Roles,
 	)
 
-	ctx.EmitEvent(api.NewEventBuilder(app.Name()).Attribute(KeyNodeRegistered, cbor.Marshal(newNode)))
+	ctx.EmitEvent(api.NewEventBuilder(app.Name()).TypedAttribute(&registry.NodeEvent{Node: newNode, IsRegistration: true}))
 
 	ctx.Commit()
 
@@ -603,7 +600,7 @@ func (app *registryApplication) unfreezeNode(
 		"node_id", node.ID,
 	)
 
-	ctx.EmitEvent(api.NewEventBuilder(app.Name()).Attribute(KeyNodeUnfrozen, cbor.Marshal(node.ID)))
+	ctx.EmitEvent(api.NewEventBuilder(app.Name()).TypedAttribute(&registry.NodeUnfrozenEvent{NodeID: node.ID}))
 
 	return nil
 }
@@ -772,7 +769,7 @@ func (app *registryApplication) registerRuntime( // nolint: gocyclo
 			"runtime", rt,
 		)
 
-		ctx.EmitEvent(api.NewEventBuilder(app.Name()).Attribute(KeyRuntimeRegistered, cbor.Marshal(rt)))
+		ctx.EmitEvent(api.NewEventBuilder(app.Name()).TypedAttribute(&registry.RuntimeEvent{Runtime: rt}))
 	}
 
 	return rt, nil