docs: update ADR 038 proposal

[egaxhaj]

For #10096

This PR introduces updates to ADR-038 for the plugin-based streaming services. These updates reflect the implementation approach taken in #13472

---

Author Checklist

All items are required. Please add a note to the item if the item is not applicable and
please add links to any relevant follow up issues.

I have...

• included the correct type prefix in the PR title
• added ! to the type prefix if API or client breaking change
• targeted the correct branch (see PR Targeting)
• provided a link to the relevant issue or specification
• followed the guidelines for building modules
• included the necessary unit and integration tests
• added a changelog entry to CHANGELOG.md
• included comments for documenting Go code
• updated the relevant documentation or specification
• reviewed "Files changed" and left comments if necessary
• confirmed all CI checks have passed

Reviewers Checklist

All items are required. Please add a note if the item is not applicable and please add
your handle next to the items reviewed if you only reviewed selected items.

I have...

• confirmed the correct type prefix in the PR title
• confirmed ! in the type prefix if API or client breaking change
• confirmed all author checklist items have been addressed
• reviewed state machine logic
• reviewed API design and naming
• reviewed documentation is accurate
• reviewed tests and test coverage
• manually tested (if applicable)

[alexanderbez]

There's a bunch of merge conflict remnants that need to be resolved

[tac0turtle]

preliminary approval. Love the new design.

[aaronc]

Is there a way to do the proto marshaling in a separate goroutine and still have the same guarantees?

Otherwise generally looks good.

[alexanderbez]

1. Missing ADR changelog entry
2. What is the interplay between the new StreamingService and the existing WriteListeners to the root-MS?
3. There is now ListenKVStore, StoreKVLIstener, StoreKVPair, ListenKVStorePair, StoreKVPairWriteListner, and KVStoreWriter...these all clobber with each other and make having a complete mental model of what everything is extremely difficult to reason about.
4. I don't understand why do we need to expose any new methods on the CacheWrapper and CacheWrapper interfaces? Intuitively, it makes sense to only expose listener logic on the root-MS which then would pass the listeners to the cache-MS when cache-wrapping it. You never use a cache-MS directly -- you always cache-wrap the root-MS.
5. In reviewing this ADR numerous times, I tried to diagram and illustrate the abstractions and inter-dependencies of what calls what, when, and how and I just find it incredibly confusing. This is exacerbated by point (3).

I would love for others to chime in here. These APIs and abstractions will have a high client/chain utilization footprint and I want to make sure we really nail down a clean UX and set of abstractions and my intuition tells me this needs more work on both.

[alexanderbez]

Actually, let me ask since I didn't think of this prior... @egaxhaj is this meant to replace or improve the existing design?

If it's meant to replace, that changes my opinion entirely. It's just not clear from the diff/PR, if your changes replace the design? If so, I actually think it's pretty clean 👍

[peterbourgon]

🙅

[yihuang]

• The ABCIListener and multistore interface changes LGTM, left two comments on the implementation details.
• I have reservations about the plugin system though, it seems pretty complicated, and I think there's alternative that can be implemented outside of sdk: #13652
• The issues about multiple service registration seems not addressed yet: #13473 (comment)
• How about I move the multistore and ABCIListener API changes into #13516, so we change the implementation together with the ADR, and leave this PR solely about the plugin system. Don't know how that affect the back-portability of #13516 though.

[yihuang]

The idea of passing change set as a whole is to avoid allocations for multiple streaming services, we can share the state listeners internally, but considering different stream services can monitor different subset of store keys, we need to filter the change set based on those keys, I guess it'll defeat the design purpose.

[alexanderbez]

For simplicity's sake, I would argue the key set filtering should exist globally, not on a per plugin/consumer based level.

[yihuang]

It's to be filtered somewhere anyway, for the out-of-process plugins, if we don't filter the change set before wire transfer, we'll need to pay the cost of transfering the whole set for each plugin, even if different plugins may only care about different stores.

[egaxhaj]

One benefit that we get with plugins over gRPC is the isolation of state streaming to a single plugin (vs the old design). Having multiple plugins filtering out state change events would bare the unnecessary cost of creating additional gRPC services just to filter out before wire transfer. There is no cost saving here. It puts additional burden on the SDK. You're also exposed to plugins overlapping on the store keys they listen. This is why I changed streaming.plugins = [] to streaming.plugin="". Do the filtering (and fan-out) downstream.

Different plugins should be reserved for exposing different parts of the system and not over the same data.

[egaxhaj]

quick update... working on addressing the latest round of questions.

[egaxhaj]

@yihuang

• I have reservations about the plugin system though, it seems pretty complicated, and I think there's alternative that can be implemented outside of sdk: consume file streamer output as reliable message queue #13652

What are you seeing as complicated? The plugin system works over gRPC. the SDK uses gRPC so nothing new here. The plugin system goes one step further and makes it easer for you to implement plugins in Go. Take a look at the plugin-go examples. There are three steps involved.

1. implement the service interface (ABCIListeneer). See file.go
2. Compile the Go plugin

go build -o streaming/plugins/abci/v1/examples/plugin-go/file streaming/plugins/abci/v1/examples/plugin-go/file.go

3. export it so it can be loaded

export COSMOS_SDK_ABCI_V1=.../streaming/plugins/abci/v1/examples/plugin-go/file

COSMOS_SDK is a prefix here and ABCI_V1 is the plugin name streaming.abci.plugin = "abci_v1"

For non Go plugins you implement the gRPC server. See plugin-python examples.

checkout egaxhaj/adr-038-grpc2 from provenance-io/cosmos-sdk and swap between plugin Impls

make test-sim-nondeterminism-streaming

# Go - file (writes to ~/)
export COSMOS_SDK_ABCI_V1=<path to sdk>/cosmos-sdk/streaming/plugins/abci/v1/examples/plugin-go/file

# python - file (writes to ~/)
export COSMOS_SDK_ABCI_V1=python3 <path to sdk>/cosmos-sdk/streaming/plugins/abci/v1/examples/plugin-python/file.py

# python - Kafka
export COSMOS_SDK_ABCI_V1=python3 <path to sdk>/cosmos-sdk/streaming/plugins/abci/v1/examples/plugin-python/kafka.py

Dependencies: python3 -m pip install confluent-kafka grpcio-tools grpcio-health-checking

---

• The issues about multiple service registration seems not addressed yet: #13473 (comment)

Addressed.

---

• How about I move the multistore and ABCIListener API changes into fix: state listener observe writes at wrong time #13516, so we change the implementation together with the ADR, and leave this PR solely about the plugin system. Don't know how that affect the back-portability of fix: state listener observe writes at wrong time #13516 though.

Let's keep them separate. Once #13516 is merged I can update the ADR as needed.

[yihuang]

The issues about multiple service registration seems not addressed yet: #13473 (comment)
Addressed.

I'm confused, I see you updated the code to register multiple plugins, what I mean is the current StreamingService interface, it's ABCIListener now, and we can't register multiple ones of those in the new proposal, right?
out-of-process plugins don't work for our case as I mentioned here, we still need the registration of custom ABCIListener(StreamingService).

[egaxhaj]

The issues about multiple service registration seems not addressed yet: #13473 (comment)
Addressed.

I'm confused, I see you updated the code to register multiple plugins, what I mean is the current StreamingService interface, it's ABCIListener now, and we can't register multiple ones of those in the new proposal, right? out-of-process plugins don't work for our case as I mentioned here, we still need the registration of custom ABCIListener(StreamingService).

StreamingService went away because there's no longer a need for the additional methods it was providing. versiondb streaming service is not using them either. It only cares about ListenBeginBlock and ListenCommit

Change versiondb StreamingService to implement ABCIListener.

var baseapp.ABCIListener = &StreamingService{}

https://github.com/crypto-org-chain/cronos/blob/936cc905f810a4381584da68418dad33099f2279/versiondb/streaming_service.go#L15

Registration is under your control. Continue to do what you're already doing (don't use the plugin system registration loop in this proposal).

# app.go

...

streamers := cast.ToString(appOpts.Get("streaming.abci.plugin"))
if strings.Contains(streamers, "file") {
  ...
}

if strings.Contains(streamers, "versions") {
  service := versiondb.NewStreamingService(versionDB, exposeStoreKeys)
  bApp.SetStreamingService(service)
  ...
}

*I need to put back SetStreamingService()

[streaming]

[streaming.abci]

plugin="file,versions"
keys=[]

# in your case you ignore it.
stop-node-on-err=false

After looking at versiondb, I understand now why you asked to not run a goroutine when stop-node-on-err=false (ref: #13473 (comment)). We can cover that use case by adding streaming.abci.async=true|false. When async=false and stop-node-on-err=false it will run without a goroutine.

edit: I'll make the updates to the ADR and post today.

edit: @yihuang I've updated the ADR to support your in-process service. You'll need to make the modifications I mentioned above but nothing major.

Registration post merge:

# app.go

pluginKey := fmt.Sprintf("%s.%s.%s", baseapp.StreamingTomlKey, baseapp.StreamingABCITomlKey, baseapp.StreamingABCIPluginTomlKey)
streamers := cast.ToString(appOpts.Get(pluginKey))
if strings.Contains(streamers, "versiondb") {
    ...
    service := versiondb.NewStreamingService(versionDB)
    bApp.SetStreamingService(service)
    bApp.cms.AddListeners(exposeStoreKeys)
    ...
}

[egaxhaj]

@alexanderbez @kocubinski @tac0turtle - I made updates that continues to support @yihuang in-process use case while still moving forward with the plugin system. Can you take another look at the ADR and if all looks good merge it?

[egaxhaj]

@tac0turtle This has three approval. Can we merge it and move onto reviewing the the implementation #14207?

[alexanderbez]

@yihuang are you OK with merging this PR? I noticed you're still requesting changes/review.

[egaxhaj]

@yihuang any objections to moving forward and merging this PR?

[yihuang]

LGTM.
Although I think different listeners should have their own async/stopNodeOnErr configurations, but it's not a blocker I guess.