feat!(Streams): Support propagating Unfolds; remove StreamResult #264

bartelink · 2024-07-15T15:46:04Z

Enhances the pipeline to support propagation of Unfold events through the pipeline:

EquinoxSystemTextJsonParser.eventsAndUnfoldsWhereStream extracts the unfolds in addition to the events
StreamSpan.merge ensures we only retain the most recent ones as inputs are coalesced
Handlers are presented any unfolds at the end of the span (assuming they fit inside the limit)
Unfolds will always have an IsUnfold = true and Index = version, where version is the Index of the last 'real' event +1
For Sync, EventStoreDbSink and CosmosStoreSink, the maxBytes limit will include/exclude either all or none of the unfolds
Retries pick up the most recent unfold

As before, given a Propulsion.Sink.Event[], you compute:

the index of the first event using Propulsion.Sink.Event.index
the index of the next event using Propulsion.Sink.Event.nextIndex (even if there are unfolds at the end of the sequence)

Note that the following assumptions no longer hold if you have unfolds in an Event[]:

nextIndex <-> index events + events.Length (renamed to next as intentional breaking change)
actual events (not unfolds) count <-> events.Length

What is the use case? If you're running a ChangeFeedProcessor from an Equinox.CosmosStore, the changefeed includes:

snapshots (which you could hydrate the state from directly without reading the store using Equinox)
RollingState state as held in the tip

Being able to route the Unfolds to the Handler alongside normal events enables:

complete copying of the content of a store in a single traversal via (propulsion sync cosmos or propulsion sync file cosmos)
being able to run a sync job (or generic event handler) on a rolling state category
being able to avoid Equinox reads of the source streams in the bulk of cases (as you can hydrate from the Unfolds instead of loading the state) using features in feat(CosmosStore.Linq): Indexed Queries equinox#450

bartelink · 2024-07-15T17:20:30Z

src/Propulsion.CosmosStore/CosmosStorePruner.fs

@@ -29,7 +29,7 @@ module Pruner =
        let res = if deleted = 0 && deferred = 0 then Nop span.Length else Ok (deleted, deferred)
        // For case where we discover events have already been deleted beyond our requested position, signal to reader to drop events
        let writePos = max trimmedPos (untilIndex + 1L)
-        return struct (writePos, res) }
+        return struct (res, writePos) }


for alignment with res, events with equinox result tuples

bartelink · 2024-07-15T17:21:15Z

src/Propulsion.CosmosStore/CosmosStorePruner.fs

@@ -75,8 +75,8 @@ type CosmosStorePruner =
 #endif
            let interpret _stream span =
                let metrics = StreamSpan.metrics Event.storedSize span
-                struct (metrics, span)
-            Dispatcher.Concurrent<_, _, _, _>.Create(maxConcurrentStreams, interpret, Pruner.handle pruneUntil, (fun _ r -> r))
+                struct (span, metrics)


any extra stuff layered on goes after the primary data in the tuple - no idea how/why I ended up having it backwards, but these signatures are undergoing change as part of putting the pos after the result anyway

bartelink · 2024-07-15T17:22:09Z

src/Propulsion.CosmosStore/CosmosStoreSink.fs

-            log.Debug("Writing {s}@{i}x{n}", stream, span[0].Index, span.Length)
+            let i = StreamSpan.index span
+            let n = StreamSpan.nextIndex span
+            span |> Seq.iter (fun x -> if x.IsUnfold then invalidOp "CosmosStore3 does not [yet] support ingesting unfolds")


prep for #263

nordfjord · 2024-07-15T19:43:16Z

src/Propulsion.CosmosStore/EquinoxSystemTextJsonParser.fs

-    /// Attempts to parse a Document/Item from the Store
+                                              correlationId = x.correlationId, causationId = x.causationId, isUnfold = isUnfold)
+        let events = batch.e |> Seq.mapi (fun offset -> gen false (batch.i + int64 offset))
+        // an Unfold won't have a corr/cause id, but that's OK - can't use Tip type as don't want to expand compressed form etc


v5 wishlist: drop correlation/causation as first class things 😇

yes, it's a misfeature that's one of the few white elephants in Equinox, but it does happen to be surfaced as top level indexable value in Equinox.CosmosStore
AFAIK nobody is actually leaning on it, but I could be wrong, so there'd conceptually have to be a placeholder task to provide a way to post process IEventData -> Event mapping in both directions.

nordfjord · 2024-07-15T19:45:16Z

src/Propulsion.CosmosStore/EquinoxSystemTextJsonParser.fs

+        tryParseEquinoxBatchOrTip streamFilter jsonDocument
+        |> ValueOption.map (fun struct (s, xs, u) -> enumEquinoxCosmosBatchOrTip (if withUnfolds then u else ValueNone) xs |> Seq.map (fun x -> s, x))


this'd read better as a plain 'ole match

It didn't but I left it!

nordfjord · 2024-07-15T19:46:46Z

src/Propulsion.Feed/PeriodicSource.fs

@@ -38,7 +38,7 @@ module private TimelineEvent =
        fun (i, x: FsCodec.IEventData<_>, context: obj) ->
            if i > DateTimeOffsetPosition.factor then invalidArg (nameof i) $"Index may not exceed %d{DateTimeOffsetPosition.factor}"
            FsCodec.Core.TimelineEvent.Create(
-                baseIndex + i, x.EventType, x.Data, x.Meta, x.EventId, x.CorrelationId, x.CausationId, x.Timestamp, isUnfold = true, context = context)


no, it's intentional - this first/only use of the flag was to convey that these were ephemeral events, not backed by a store
but nothing leans on that
the change in this PR as a whole is that IsUnfold events now get dropped and replaced whenever a new set comes through (similar to how cosmos changefeed will deliver only the latest state)
But that would be bad for periodic source, which generates a set of sequntial events to represent the full state of the input at a point in time (and then re-runs the search with a new prefix when the period resets)

src/Propulsion/Sinks.fs

src/Propulsion/Streams.fs

nordfjord

Not picking up on anything too bad, but this would be a new major version since the user facing API seems to be changing

bartelink · 2024-07-15T20:45:33Z

but this would be a new major version since the user facing API seems to be changing

You are not wrong of course... But I have some buts:

This handler signature was only introduced in the V3 timeframe, and I believe it to have been a misdesign - you want calculating the next version (or having it flow from what the sycing is going into) be foremost in people's minds

For a V4, I think Propulsion.Feed and Propulsion should be merged as lots of modules depend on Propulsion.Feed and it in turn depends on Propulsion

What if I promised to mint a V4 as soon as Equinox.CosmosStore 4.1.0 is final ? (and kill Propulsion.CosmosStore3 the minute that's done)

Other things on the list that I'd like to do at some point (which don't change the V4 API but simplify the whole deal):

align Propulsion.Kafka's Source mechanism with how all the other sources work (it has lots of ugly parallel implementations that are pretty close to what Propulsion.Feed generalizes for everything else
other stuff in V3 checklist #116 like folding in the EventStore striped reading stuff into Propulsion.EventStoreDb

feat!(Streams): Support propagating Unfolds

c4c09df

bartelink mentioned this pull request Jul 15, 2024

feat(Cosmos, propulsion sync): propagate unfolds #263

Merged

bartelink commented Jul 15, 2024

View reviewed changes

nordfjord reviewed Jul 15, 2024

View reviewed changes

src/Propulsion/Sinks.fs Show resolved Hide resolved

nordfjord reviewed Jul 15, 2024

View reviewed changes

src/Propulsion/Streams.fs Outdated Show resolved Hide resolved

nordfjord approved these changes Jul 15, 2024

View reviewed changes

bartelink added 4 commits July 16, 2024 07:27

Address review comments

a112eb7

Defer change to next PR

207269f

Rewordings

a46e104

Tidy

a9261d8

bartelink changed the title ~~feat!(Streams): Support propagating Unfolds~~ feat!(Streams): Support propagating Unfolds; remove StreamResult Jul 16, 2024

bartelink added 4 commits July 16, 2024 13:04

Polish maxItems

d32a4a5

Tidy/fix slice algo

bcacc6c

Format/style changes

642d189

Renames: nextIndex->next

50d6eb8

bartelink force-pushed the remove-streamresult branch from 68afadf to 50d6eb8 Compare August 1, 2024 15:21

Batch Ingestion/Progress extensions/fixes

b375d90

bartelink merged commit 40c9ab4 into master Aug 1, 2024
6 checks passed

bartelink deleted the remove-streamresult branch August 1, 2024 16:01

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat!(Streams): Support propagating Unfolds; remove StreamResult #264

feat!(Streams): Support propagating Unfolds; remove StreamResult #264

bartelink commented Jul 15, 2024 •

edited

Loading

bartelink Jul 15, 2024

bartelink Jul 15, 2024

bartelink Jul 15, 2024

nordfjord Jul 15, 2024

bartelink Jul 15, 2024

nordfjord Jul 15, 2024

bartelink Jul 16, 2024

nordfjord Jul 15, 2024

bartelink Jul 15, 2024

nordfjord left a comment

bartelink commented Jul 15, 2024

		tryParseEquinoxBatchOrTip streamFilter jsonDocument
		\|> ValueOption.map (fun struct (s, xs, u) -> enumEquinoxCosmosBatchOrTip (if withUnfolds then u else ValueNone) xs \|> Seq.map (fun x -> s, x))

feat!(Streams): Support propagating Unfolds; remove StreamResult #264

feat!(Streams): Support propagating Unfolds; remove StreamResult #264

Conversation

bartelink commented Jul 15, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nordfjord left a comment

Choose a reason for hiding this comment

bartelink commented Jul 15, 2024

bartelink commented Jul 15, 2024 •

edited

Loading