fix(stf): fixes to make init genesis pass (#21088)

Co-authored-by: marbar3778 <[email protected]>
(cherry picked from commit b05bb26)

# Conflicts:
#	runtime/v2/builder.go
#	schema/appdata/async.go
#	schema/appdata/mux_test.go
#	schema/module_schema_test.go
#	server/v2/store/commands.go

runtime/v2/builder.go

@@ -119,6 +119,33 @@ func (a *AppBuilder[T]) Build(opts ...AppBuilderOption[T]) (*App[T], error) {
 	}
 	a.app.stf = stf
 
+<<<<<<< HEAD
+=======
+	v := a.viper
+	home := v.GetString(FlagHome)
+
+	storeOpts := rootstore.DefaultStoreOptions()
+	if s := v.Sub("store.options"); s != nil {
+		if err := s.Unmarshal(&storeOpts); err != nil {
+			return nil, fmt.Errorf("failed to store options: %w", err)
+		}
+	}
+
+	scRawDb, err := db.NewDB(db.DBType(v.GetString("store.app-db-backend")), "application", filepath.Join(home, "data"), nil)
+	if err != nil {
+		panic(err)
+	}
+
+	storeOptions := &rootstore.FactoryOptions{
+		Logger:    a.app.logger,
+		RootDir:   home,
+		Options:   storeOpts,
+		StoreKeys: append(a.app.storeKeys, "stf"),
+		SCRawDB:   scRawDb,
+	}
+	a.storeOptions = storeOptions
+
+>>>>>>> b05bb2601 (fix(stf): fixes to make init genesis pass (#21088))
 	rs, err := rootstore.CreateRootStore(a.storeOptions)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create root store: %w", err)

runtime/v2/manager.go

@@ -166,7 +166,7 @@ func (m *MM[T]) InitGenesisJSON(
 		case appmodulev2.HasGenesis:
 			m.logger.Debug("running initialization for module", "module", moduleName)
 			if err := module.InitGenesis(ctx, genesisData[moduleName]); err != nil {
-				return err
+				return fmt.Errorf("init module %s: %w", moduleName, err)
 			}
 		case appmodulev2.HasABCIGenesis:
 			m.logger.Debug("running initialization for module", "module", moduleName)

schema/appdata/async.go

@@ -0,0 +1,162 @@
+package appdata
+
+import (
+	"context"
+	"sync"
+)
+
+// AsyncListenerOptions are options for async listeners and listener mux's.
+type AsyncListenerOptions struct {
+	// Context is the context whose Done() channel listeners use will use to listen for completion to close their
+	// goroutine. If it is nil, then context.Background() will be used and goroutines may be leaked.
+	Context context.Context
+
+	// BufferSize is the buffer size of the channels to use. It defaults to 0.
+	BufferSize int
+
+	// DoneWaitGroup is an optional wait-group that listener goroutines will notify via Add(1) when they are started
+	// and Done() after they are canceled and completed.
+	DoneWaitGroup *sync.WaitGroup
+}
+
+// AsyncListenerMux returns a listener that forwards received events to all the provided listeners asynchronously
+// with each listener processing in a separate go routine. All callbacks in the returned listener will return nil
+// except for Commit which will return an error or nil once all listeners have processed the commit. The context
+// is used to signal that the listeners should stop listening and return. bufferSize is the size of the buffer for the
+// channels used to send events to the listeners.
+func AsyncListenerMux(opts AsyncListenerOptions, listeners ...Listener) Listener {
+	asyncListeners := make([]Listener, len(listeners))
+	commitChans := make([]chan error, len(listeners))
+	for i, l := range listeners {
+		commitChan := make(chan error)
+		commitChans[i] = commitChan
+		asyncListeners[i] = AsyncListener(opts, commitChan, l)
+	}
+	mux := ListenerMux(asyncListeners...)
+	muxCommit := mux.Commit
+	mux.Commit = func(data CommitData) error {
+		if muxCommit != nil {
+			err := muxCommit(data)
+			if err != nil {
+				return err
+			}
+		}
+
+		for _, commitChan := range commitChans {
+			err := <-commitChan
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+
+	return mux
+}
+
+// AsyncListener returns a listener that forwards received events to the provided listener listening in asynchronously
+// in a separate go routine. The listener that is returned will return nil for all methods including Commit and
+// an error or nil will only be returned in commitChan once the sender has sent commit and the receiving listener has
+// processed it. Thus commitChan can be used as a synchronization and error checking mechanism. The go routine
+// that is being used for listening will exit when context.Done() returns and no more events will be received by the listener.
+// bufferSize is the size of the buffer for the channel that is used to send events to the listener.
+// Instead of using AsyncListener directly, it is recommended to use AsyncListenerMux which does coordination directly
+// via its Commit callback.
+func AsyncListener(opts AsyncListenerOptions, commitChan chan<- error, listener Listener) Listener {
+	packetChan := make(chan Packet, opts.BufferSize)
+	res := Listener{}
+	ctx := opts.Context
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	done := ctx.Done()
+
+	go func() {
+		if opts.DoneWaitGroup != nil {
+			opts.DoneWaitGroup.Add(1)
+		}
+
+		var err error
+		for {
+			select {
+			case packet := <-packetChan:
+				if err != nil {
+					// if we have an error, don't process any more packets
+					// and return the error and finish when it's time to commit
+					if _, ok := packet.(CommitData); ok {
+						commitChan <- err
+						return
+					}
+				} else {
+					// process the packet
+					err = listener.SendPacket(packet)
+					// if it's a commit
+					if _, ok := packet.(CommitData); ok {
+						commitChan <- err
+						if err != nil {
+							return
+						}
+					}
+				}
+
+			case <-done:
+				close(packetChan)
+				if opts.DoneWaitGroup != nil {
+					opts.DoneWaitGroup.Done()
+				}
+				return
+			}
+		}
+	}()
+
+	if listener.InitializeModuleData != nil {
+		res.InitializeModuleData = func(data ModuleInitializationData) error {
+			packetChan <- data
+			return nil
+		}
+	}
+
+	if listener.StartBlock != nil {
+		res.StartBlock = func(data StartBlockData) error {
+			packetChan <- data
+			return nil
+		}
+	}
+
+	if listener.OnTx != nil {
+		res.OnTx = func(data TxData) error {
+			packetChan <- data
+			return nil
+		}
+	}
+
+	if listener.OnEvent != nil {
+		res.OnEvent = func(data EventData) error {
+			packetChan <- data
+			return nil
+		}
+	}
+
+	if listener.OnKVPair != nil {
+		res.OnKVPair = func(data KVPairData) error {
+			packetChan <- data
+			return nil
+		}
+	}
+
+	if listener.OnObjectUpdate != nil {
+		res.OnObjectUpdate = func(data ObjectUpdateData) error {
+			packetChan <- data
+			return nil
+		}
+	}
+
+	if listener.Commit != nil {
+		res.Commit = func(data CommitData) error {
+			packetChan <- data
+			return nil
+		}
+	}
+
+	return res
+}

schema/appdata/mux_test.go

@@ -0,0 +1,132 @@
+package appdata
+
+import (
+	"fmt"
+	"testing"
+)
+
+func TestListenerMux(t *testing.T) {
+	t.Run("empty", func(t *testing.T) {
+		listener := ListenerMux(Listener{}, Listener{})
+
+		if listener.InitializeModuleData != nil {
+			t.Error("expected nil")
+		}
+		if listener.StartBlock != nil {
+			t.Error("expected nil")
+		}
+		if listener.OnTx != nil {
+			t.Error("expected nil")
+		}
+		if listener.OnEvent != nil {
+			t.Error("expected nil")
+		}
+		if listener.OnKVPair != nil {
+			t.Error("expected nil")
+		}
+		if listener.OnObjectUpdate != nil {
+			t.Error("expected nil")
+		}
+		if listener.Commit != nil {
+			t.Error("expected nil")
+		}
+	})
+
+	t.Run("all called once", func(t *testing.T) {
+		var calls []string
+		onCall := func(name string, i int, _ Packet) {
+			calls = append(calls, fmt.Sprintf("%s %d", name, i))
+		}
+
+		res := ListenerMux(callCollector(1, onCall), callCollector(2, onCall))
+
+		callAllCallbacksOnces(t, res)
+
+		checkExpectedCallOrder(t, calls, []string{
+			"InitializeModuleData 1",
+			"InitializeModuleData 2",
+			"StartBlock 1",
+			"StartBlock 2",
+			"OnTx 1",
+			"OnTx 2",
+			"OnEvent 1",
+			"OnEvent 2",
+			"OnKVPair 1",
+			"OnKVPair 2",
+			"OnObjectUpdate 1",
+			"OnObjectUpdate 2",
+			"Commit 1",
+			"Commit 2",
+		})
+	})
+}
+
+func callAllCallbacksOnces(t *testing.T, listener Listener) {
+	if err := listener.InitializeModuleData(ModuleInitializationData{}); err != nil {
+		t.Error(err)
+	}
+	if err := listener.StartBlock(StartBlockData{}); err != nil {
+		t.Error(err)
+	}
+	if err := listener.OnTx(TxData{}); err != nil {
+		t.Error(err)
+	}
+	if err := listener.OnEvent(EventData{}); err != nil {
+		t.Error(err)
+	}
+	if err := listener.OnKVPair(KVPairData{}); err != nil {
+		t.Error(err)
+	}
+	if err := listener.OnObjectUpdate(ObjectUpdateData{}); err != nil {
+		t.Error(err)
+	}
+	if err := listener.Commit(CommitData{}); err != nil {
+		t.Error(err)
+	}
+}
+
+func callCollector(i int, onCall func(string, int, Packet)) Listener {
+	return Listener{
+		InitializeModuleData: func(ModuleInitializationData) error {
+			onCall("InitializeModuleData", i, nil)
+			return nil
+		},
+		StartBlock: func(StartBlockData) error {
+			onCall("StartBlock", i, nil)
+			return nil
+		},
+		OnTx: func(TxData) error {
+			onCall("OnTx", i, nil)
+			return nil
+		},
+		OnEvent: func(EventData) error {
+			onCall("OnEvent", i, nil)
+			return nil
+		},
+		OnKVPair: func(KVPairData) error {
+			onCall("OnKVPair", i, nil)
+			return nil
+		},
+		OnObjectUpdate: func(ObjectUpdateData) error {
+			onCall("OnObjectUpdate", i, nil)
+			return nil
+		},
+		Commit: func(CommitData) error {
+			onCall("Commit", i, nil)
+			return nil
+		},
+	}
+}
+
+func checkExpectedCallOrder(t *testing.T, actual, expected []string) {
+	t.Helper()
+	if len(actual) != len(expected) {
+		t.Fatalf("expected %d calls, got %d", len(expected), len(actual))
+	}
+
+	for i := range actual {
+		if actual[i] != expected[i] {
+			t.Errorf("expected %q, got %q", expected[i], actual[i])
+		}
+	}
+}