feat: Optional async decoupling for secondary writes and reworked E2E metric assertions #182
Conversation
…ondary insertions
…ondary insertions
There was a problem hiding this comment.
I like the refactor! Think there are ways to make the API for secondary struct easier to understand and use though..! I'm not a big fan of async APIs. See my reasoning here: https://www.notion.so/eigen-labs/Bls-Agg-Service-2-0-synchronous-API-98abef46040a48fc8d044e7de0781839
README.md
Outdated
| Unit tests can be ran via invoking `make test`. Please make sure to have all test containers downloaded locally before running via: | ||
| ``` | ||
| docker pull redis | ||
| docker pull minio | ||
| ``` |
There was a problem hiding this comment.
is this really needed? I would think testcontainer would still pull the images when attempting to run them if they are not present locally?
…via metrics - cleanups
…via metrics - refactors and lints
…via metrics - refactors and lints
…via metrics - refactors and lints
…via metrics - ensure thread safety for secondary stores
…via metrics - use in memory metrics
…via metrics - add concurrency flag
…via metrics - fmt
epociask
changed the title
epociask
changed the title
store/secondary.go
Outdated
| type ISecondary interface { | ||
| AsyncEntry() bool | ||
| Enabled() bool | ||
| Topic() chan<- PutNotify | ||
| CachingEnabled() bool | ||
| FallbackEnabled() bool | ||
| HandleRedundantWrites(ctx context.Context, commitment []byte, value []byte) error | ||
| MultiSourceRead(context.Context, []byte, bool, func([]byte, []byte) error) ([]byte, error) | ||
| WriteSubscriptionLoop(ctx context.Context) | ||
| } |
There was a problem hiding this comment.
This probably works. I personally feel like we're getting wayyyy to complex way too early, for something that's super simple: we only have s3 and redis options, but now we have all these indirection layers, caching vs fallback, interfaces over interfaces. Makes it very hard to understand the code, review it, find bugs, benchmark/profile and optimize, etc. Let's merge this, but I'd like to try to keep the complexity lower in the future, or at least think through ways of simplifying the architecture.
…enchmarks, code comments
…ociask--chore-reabstract-router
…ociask--chore-reabstract-router
…ociask--chore-reabstract-router
e2e/setup.go
Outdated
| @@ -113,6 +111,7 @@ type Cfg struct { | |||
| UseS3Caching bool | |||
| UseRedisCaching bool | |||
| UseS3Fallback bool | |||
| WriteThreadCount int | |||
There was a problem hiding this comment.
prob makes the comment above ("at most one of the below options should be true") not true anymore. Maybe update the comment to say "below 3/4 options"
e2e/setup.go
Outdated
| // StringOrByte ... constraint that generic type is either "string" or "[]byte" | ||
| type StringOrByte interface { | ||
| string | []byte | ||
| } | ||
|
|
||
| func randStr(n int) string { | ||
| var letterRunes = []rune("abcdefghijklmnopqrstuvwxyz") | ||
| b := make([]rune, n) | ||
| for i := range b { | ||
| b[i] = letterRunes[rand.Intn(len(letterRunes))] | ||
| } | ||
| return string(b) | ||
| } | ||
|
|
||
| // Rand generates either a random string or byte slice depending on the type T | ||
| func Rand[T StringOrByte](length int) T { | ||
| str := randStr(length) | ||
|
|
||
| // Use type switch to return the correct type | ||
| var result any | ||
| switch any(new(T)).(type) { | ||
| case *string: | ||
| result = str | ||
| case *[]byte: | ||
| result = []byte(str) | ||
| } | ||
|
|
||
| return result.(T) | ||
| } |
There was a problem hiding this comment.
Think this is overcomplicating things and not necessary. Generics typically when behavior is the same, but types are different. Here we just want 2 different things. Why not just have 2 separate functions RandByte and RandString? Super straightforward, follows stdlib pattern. At your call site if I see the generic I'm not sure what's happening so I need to click through and then parse this complex function, for something that should be completely trivial and uninteresting.
| @@ -33,6 +33,7 @@ func withMetrics( | |||
| // we assume that every route will set the status header | |||
| versionByte, err := parseVersionByte(w, r) | |||
| if err != nil { | |||
| recordDur(w.Header().Get("status"), string(mode), string(versionByte)) | |||
There was a problem hiding this comment.
This w.Header().Get("status") thing is buggy, I have an issue about this somewhere. But let's keep it like this for now and fix it in another PR.
Fixes Issue
Related to #164 but doesn't fix it.
Fixes #113 and #96
Changes proposed
Benchmarking
As expected, latency decreases when using async writes to secondary storage.
0 threads:
5 threads:
10 threads:
Note to reviewers