svs: return instance state as binary

Author: pulsejet

dv/dv/router.go

@@ -343,7 +343,8 @@ func (dv *Router) createPrefixTable() {
 	dv.pfxSubs = make(map[uint64]enc.Name)
 
 	// SVS delivery agent
-	dv.pfxSvs = ndn_sync.NewSvsALO(ndn_sync.SvsAloOpts{
+	var err error
+	dv.pfxSvs, err = ndn_sync.NewSvsALO(ndn_sync.SvsAloOpts{
 		Name: dv.config.RouterDataPrefix(),
 		Svs: ndn_sync.SvSyncOpts{
 			Client:      dv.client,
@@ -358,6 +359,9 @@ func (dv *Router) createPrefixTable() {
 			Threshold: PrefixSnapThreshold,
 		},
 	})
+	if err != nil {
+		panic(err)
+	}
 
 	// Local prefix table
 	dv.pfx = table.NewPrefixTable(dv.config, func(w enc.Wire) {

std/examples/svs/alo-history/main.go

@@ -10,7 +10,6 @@ import (
 	"github.com/named-data/ndnd/std/engine"
 	"github.com/named-data/ndnd/std/log"
 	"github.com/named-data/ndnd/std/ndn"
-	spec_svs "github.com/named-data/ndnd/std/ndn/svs/v3"
 	"github.com/named-data/ndnd/std/ndn/svs_ps"
 	"github.com/named-data/ndnd/std/object"
 	ndn_sync "github.com/named-data/ndnd/std/sync"
@@ -74,7 +73,7 @@ func main() {
 	defer client.Stop()
 
 	// Create a new SVS ALO instance
-	svsalo = ndn_sync.NewSvsALO(ndn_sync.SvsAloOpts{
+	svsalo, err = ndn_sync.NewSvsALO(ndn_sync.SvsAloOpts{
 		// Name is the name of the node
 		Name: name,
 
@@ -95,6 +94,9 @@ func main() {
 			Threshold: 10,
 		},
 	})
+	if err != nil {
+		panic(err)
+	}
 
 	// OnError gets called when we get an error from the SVS ALO instance.
 	svsalo.SetOnError(func(err error) {
@@ -190,30 +192,21 @@ func publish(content []byte) {
 	commitState(state)
 }
 
-func commitState(state *spec_svs.InstanceState) {
+func commitState(state enc.Wire) {
 	// Once a publication is processed, ideally the application should persist
 	// it's own state and the state of the Sync group *atomically*.
 	//
 	// Applications can use their own data structures to store the state.
-	// In this example, we use the object store to persist the state.
-	store.Put(group, 0, state.Encode().Join())
+	// In this example, we use the NDN object store to persist the state.
+	store.Put(group, 0, state.Join())
 }
 
-func readState() *spec_svs.InstanceState {
+func readState() enc.Wire {
 	// Read the state from the object store
 	// See commitState for more information
 	stateWire, err := store.Get(group, false)
 	if err != nil {
-		log.Error(nil, "Unable to get state from object store", "err", err)
-		os.Exit(1)
-	}
-	if stateWire != nil {
-		state, err := spec_svs.ParseInstanceState(enc.NewBufferView(stateWire), true)
-		if err != nil {
-			log.Error(nil, "Unable to parse state from object store", "err", err)
-			os.Exit(1)
-		}
-		return state
+		panic("unable to get state (store is broken)")
 	}
-	return nil
+	return enc.Wire{stateWire}
 }

std/examples/svs/alo-latest/main.go

@@ -71,7 +71,7 @@ func main() {
 	msgSize := 0
 
 	// Create a new SVS ALO instance
-	svsalo := ndn_sync.NewSvsALO(ndn_sync.SvsAloOpts{
+	svsalo, err := ndn_sync.NewSvsALO(ndn_sync.SvsAloOpts{
 		// Name is the name of the node
 		Name: name,
 
@@ -102,6 +102,9 @@ func main() {
 			Threshold: 10,
 		},
 	})
+	if err != nil {
+		panic(err)
+	}
 
 	// Set of publishers that we are subscribed to.
 	// The value is the number of errors received for the publisher.

std/sync/svs_alo.go

@@ -49,15 +49,15 @@ type SvsAloOpts struct {
 	// Snapshot is the snapshot strategy.
 	Snapshot Snapshot
 	// InitialState is the initial state of the instance.
-	InitialState *spec_svs.InstanceState
+	InitialState enc.Wire
 
 	// MaxPipelineSize is the number of objects to fetch
 	// concurrently for a single publisher (default 10)
 	MaxPipelineSize uint64
 }
 
 // NewSvsALO creates a new SvsALO instance.
-func NewSvsALO(opts SvsAloOpts) *SvsALO {
+func NewSvsALO(opts SvsAloOpts) (*SvsALO, error) {
 	if len(opts.Name) == 0 {
 		panic("Name is required")
 	}
@@ -89,21 +89,8 @@ func NewSvsALO(opts SvsAloOpts) *SvsALO {
 	// Read initial state if provided.
 	s.opts.Svs.OnUpdate = s.onSvsUpdate
 	if s.opts.InitialState != nil {
-		if !s.opts.InitialState.Name.Equal(s.opts.Name) {
-			panic("Name mismatch in provided initial state")
-		}
-		s.opts.Svs.BootTime = s.opts.InitialState.BootstrapTime
-		s.opts.Svs.InitialState = s.opts.InitialState.StateVector
-
-		for _, entry := range s.opts.InitialState.StateVector.Entries {
-			hash := entry.Name.TlvStr()
-			for _, seqEntry := range entry.SeqNoEntries {
-				s.state.Set(hash, seqEntry.BootstrapTime, svsDataState{
-					Known:   seqEntry.SeqNo,
-					Latest:  seqEntry.SeqNo,
-					Pending: seqEntry.SeqNo,
-				})
-			}
+		if err := s.parseInstanceState(s.opts.InitialState); err != nil {
+			return nil, err
 		}
 	}
 
@@ -131,7 +118,7 @@ func NewSvsALO(opts SvsAloOpts) *SvsALO {
 		})
 	}
 
-	return s
+	return s, nil
 }
 
 // String is the log identifier.
@@ -194,7 +181,7 @@ func (s *SvsALO) SetOnPublisher(callback func(enc.Name)) {
 }
 
 // Publish sends a message to the group
-func (s *SvsALO) Publish(content enc.Wire) (enc.Name, *spec_svs.InstanceState, error) {
+func (s *SvsALO) Publish(content enc.Wire) (name enc.Name, state enc.Wire, err error) {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 

std/sync/svs_alo_data.go

@@ -1,6 +1,7 @@
 package sync
 
 import (
+	"fmt"
 	"slices"
 	"time"
 
@@ -34,7 +35,7 @@ func (s *SvsALO) objectName(node enc.Name, boot uint64, seq uint64) enc.Name {
 		WithVersion(enc.VersionImmutable)
 }
 
-func (s *SvsALO) produceObject(content enc.Wire) (enc.Name, *spec_svs.InstanceState, error) {
+func (s *SvsALO) produceObject(content enc.Wire) (enc.Name, enc.Wire, error) {
 	// This instance owns the underlying SVS instance.
 	// So we can be sure that the sequence number does not
 	// change while we hold the lock on this instance.
@@ -231,23 +232,51 @@ func (s *SvsALO) queuePub(pub SvsPub) {
 	s.outpipe <- pub
 }
 
-// instanceState returns the current state of the instance.
-func (s *SvsALO) instanceState() *spec_svs.InstanceState {
-	stateVector := s.state.Encode(func(state svsDataState) uint64 {
-		return state.Known
-	})
+// queueError queues an error to the application.
+func (s *SvsALO) queueError(err error) {
+	select {
+	case s.errpipe <- err:
+	default:
+	}
+}
 
-	return &spec_svs.InstanceState{
+// instanceState returns the current state of the instance.
+func (s *SvsALO) instanceState() enc.Wire {
+	state := spec_svs.InstanceState{
 		Name:          s.opts.Name,
 		BootstrapTime: s.BootTime(),
-		StateVector:   stateVector,
+		StateVector: s.state.Encode(func(state svsDataState) uint64 {
+			return state.Known
+		}),
 	}
+	return state.Encode()
 }
 
-// queueError queues an error to the application.
-func (s *SvsALO) queueError(err error) {
-	select {
-	case s.errpipe <- err:
-	default:
+// parseInstanceState parses an instance state into the current state.
+// Only the constructor should call this function.
+func (s *SvsALO) parseInstanceState(wire enc.Wire) error {
+	initState, err := spec_svs.ParseInstanceState(enc.NewWireView(wire), true)
+	if err != nil {
+		return err
+	}
+
+	if !initState.Name.Equal(s.opts.Name) {
+		return fmt.Errorf("initial state name mismatch: %v != %v", initState.Name, s.opts.Name)
 	}
+
+	s.opts.Svs.BootTime = initState.BootstrapTime
+	s.opts.Svs.InitialState = initState.StateVector
+
+	for _, entry := range initState.StateVector.Entries {
+		hash := entry.Name.TlvStr()
+		for _, seqEntry := range entry.SeqNoEntries {
+			s.state.Set(hash, seqEntry.BootstrapTime, svsDataState{
+				Known:   seqEntry.SeqNo,
+				Latest:  seqEntry.SeqNo,
+				Pending: seqEntry.SeqNo,
+			})
+		}
+	}
+
+	return nil
 }

std/sync/svs_pub.go

@@ -2,7 +2,6 @@ package sync
 
 import (
 	enc "github.com/named-data/ndnd/std/encoding"
-	spec_svs "github.com/named-data/ndnd/std/ndn/svs/v3"
 )
 
 // SvsPub is the generic received data publication from SVS.
@@ -20,7 +19,7 @@ type SvsPub struct {
 	// IsSnapshot is true if this is a snapshot.
 	IsSnapshot bool
 	// State is the state after this publication is applied.
-	State *spec_svs.InstanceState
+	State enc.Wire
 
 	// subcribers is the list of subscribers.
 	subcribers []func(SvsPub)

std/utils/utils.go

@@ -39,7 +39,7 @@ func ConvertNonce(nonce []byte) (ret optional.Optional[uint32]) {
 	return ret
 }
 
-// If is the ternary operator
+// If is the ternary operator (eager evaluation)
 func If[T any](cond bool, t, f T) T {
 	if cond {
 		return t