feat: expose job operations over the C ABI

.github/workflows/ci.yaml

@@ -134,6 +134,40 @@ jobs:
       - name: examples
         run: nix run .#check-examples
 
+  haskell:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0  # v7.0.0
+        with:
+          persist-credentials: false
+
+      - uses: nixbuild/nix-quick-install-action@2c9db80fb984ceb1bcaa77cdda3fdf8cfba92035  # v34
+        with:
+          nix_version: "2.31.2"
+          nix_conf: |
+            accept-flake-config = true
+            fallback = true
+            access-tokens = github.com=${{ secrets.GITHUB_TOKEN }}
+            keep-env-derivations = true
+            keep-outputs = true
+
+      - uses: cachix/cachix-action@5f2d7c5294214f71b873db4b969586b980625e71  # v17
+        continue-on-error: true
+        with:
+          name: data-cartel-public
+          authToken: ${{ secrets.CACHIX_AUTH_TOKEN }}
+
+      - uses: nix-community/cache-nix-action@7df957e333c1e5da7721f60227dbba6d06080569  # v7
+        with:
+          primary-key: nix-${{ runner.os }}-haskell-${{ hashFiles('**/*.nix', '**/flake.lock') }}
+          restore-prefixes-first-match: nix-${{ runner.os }}-haskell-
+          gc-max-store-size-linux: 5G
+
+      - name: haskell
+        run: nix build .#checks.x86_64-linux.haskell -L
+
   hooks:
     runs-on: ubuntu-latest
     timeout-minutes: 10

Cargo.toml

@@ -43,6 +43,7 @@ cbindgen = "=0.29.2"
 ciborium = "0.2.2"
 cqrs-es = "0.5.0"
 event-sorcery = { path = "crates/event-sorcery" }
+getrandom = "0.3.4"
 serde = { version = "1.0.228", features = ["derive"] }
 serde_json = "1.0.150"
 sqlite-es = { path = "crates/sqlite-es" }

SPEC.md

@@ -264,15 +264,22 @@ boundary:
 
 ABI major version 0 has these normative resource limits:
 
-| Resource                                        |  Limit |
-| ----------------------------------------------- | -----: |
-| Encoded request buffer                          | 16 MiB |
-| Encoded response buffer, including error detail | 64 MiB |
-| CBOR container nesting depth                    |     32 |
-| One opaque domain payload                       |  1 MiB |
-| Events in one commit                            |  1,024 |
-| Items in one page or list response              |  4,096 |
-| UTF-8 error-detail text                         |  4 KiB |
+| Resource                                        |            Limit |
+| ----------------------------------------------- | ---------------: |
+| Encoded request buffer                          |           16 MiB |
+| Encoded response buffer, including error detail |           64 MiB |
+| CBOR container nesting depth                    |               32 |
+| One opaque domain payload                       |            1 MiB |
+| Events in one commit                            |            1,024 |
+| Items in one page or list response              |            4,096 |
+| UTF-8 error-detail text                         |            4 KiB |
+| UTF-8 identifier text                           |            4 KiB |
+| Store busy timeout                              | 2,147,483,647 ms |
+
+An identifier is any caller-supplied name the store persists or matches on: an
+aggregate type or id, a job id or kind, and a worker name. The busy-timeout
+bound is what the SQLite driver accepts; a larger value is refused at open
+rather than failing later as an infrastructure error.
 
 Lengths and counts MUST be checked before proportional allocation or durable
 work whenever the encoded shape makes that possible. Exceeding any limit MUST

bindings/haskell/event-sorcery.cabal

@@ -52,12 +52,11 @@ library
   exposed-modules:
     EventSorcery.Engine
     EventSorcery.Engine.Codec
-    EventSorcery.Engine.Protocol
-
-  other-modules:
-    EventSorcery.Engine.Acquisition
     EventSorcery.Engine.Internal.FFI
+    EventSorcery.Engine.Internal.Paging
+    EventSorcery.Engine.Protocol
 
+  other-modules:   EventSorcery.Engine.Acquisition
   extra-libraries: event_sorcery_ffi
   build-depends:
     , base                       >=4.18   && <5
@@ -85,6 +84,7 @@ test-suite engine-spec
   hs-source-dirs: test
   main-is:        EngineSpec.hs
   other-modules:  EventSorcery.Engine.AcquisitionSpec
+  ghc-options:    -threaded -with-rtsopts=-N
   build-depends:
     , base                       >=4.18 && <5
     , bytestring                 >=0.11 && <0.13

bindings/haskell/src/EventSorcery/Engine.hs

@@ -1,11 +1,15 @@
 module EventSorcery.Engine (
   Store,
   abiVersion,
+  checkAbiVersion,
   closeStore,
   commit,
   currentVersion,
   loadStream,
+  loadStreamPage,
+  minimumAbiMinor,
   openStore,
+  supportedAbiMajor,
 ) where
 
 import Control.Concurrent.MVar (MVar, newMVar, withMVar)
@@ -14,6 +18,7 @@ import Data.Bifunctor (first)
 import Data.Bits (shiftR, (.&.))
 import Data.ByteString (ByteString)
 import Data.ByteString qualified as ByteString
+import Data.ByteString.Unsafe (unsafeUseAsCStringLen)
 import Data.List.NonEmpty (NonEmpty)
 import Data.List.NonEmpty qualified as NonEmpty
 import Data.Text qualified as Text
@@ -23,6 +28,7 @@ import EventSorcery.Engine.Acquisition (
   acquireStore,
  )
 import EventSorcery.Engine.Codec (
+  decodeCloseStatus,
   decodeEngineError,
   decodeStoredEvents,
   encodeCommit,
@@ -41,12 +47,15 @@ import EventSorcery.Engine.Internal.FFI (
   esLoadStream,
   esOpen,
  )
+import EventSorcery.Engine.Internal.Paging (nextCursor)
 import EventSorcery.Engine.Protocol (
   AbiVersionDetail (..),
+  BindingFault (..),
   EngineError (..),
+  EngineErrorDecodeDetail (..),
   OpenOptions,
   ProposedEvent,
-  StoredEvent,
+  StoredEvent (..),
   StreamIdentity,
  )
 import Foreign.C.Types (CInt, CSize)
@@ -62,10 +71,10 @@ import Prelude (
   Eq ((==)),
   IO,
   Int,
-  Maybe,
-  Show (show),
+  Maybe (Just, Nothing),
   String,
   fromIntegral,
+  id,
   pure,
   ($),
   (&&),
@@ -85,6 +94,39 @@ abiVersion :: IO Word32
 abiVersion = esAbiVersion
 
 
+-- | The ABI major version this binding speaks.
+supportedAbiMajor :: Word32
+supportedAbiMajor = 0
+
+
+-- | The oldest ABI minor version whose calls this binding can make.
+minimumAbiMinor :: Word32
+minimumAbiMinor = 2
+
+
+-- | Decides whether a packed engine ABI version can serve this binding.
+--
+-- Minor versions are additive, so any engine at or above the floor is
+-- accepted and a newer engine does not need a new binding.
+checkAbiVersion :: Word32 -> Either EngineError ()
+checkAbiVersion version =
+  if actualMajor == supportedAbiMajor && actualMinor >= minimumAbiMinor
+    then Right ()
+    else
+      Left
+        ( AbiVersionMismatch
+            ( AbiVersionDetail
+                supportedAbiMajor
+                minimumAbiMinor
+                actualMajor
+                actualMinor
+            )
+        )
+  where
+    actualMajor = version `shiftR` 16
+    actualMinor = version .&. 0xffff
+
+
 openStore :: OpenOptions -> IO (Either EngineError Store)
 openStore options = do
   version <- abiVersion
@@ -98,15 +140,30 @@ closeStore (Store owner gate) =
   withMVar gate $ \() ->
     withForeignPtr owner $ \cell -> do
       status <- esClose cell
-      pure (statusWithoutDetail status)
+      pure (decodeCloseStatus (fromIntegral status))
 
 
+-- | Reads every event after the cursor, walking as many pages as it takes.
+--
+-- The engine bounds one page by an event count and by a byte budget, so a
+-- short page is not by itself the end of the stream; the stream ends at the
+-- first empty page. Callers that want to handle one page at a time -- to keep
+-- a long stream out of memory, say -- use 'loadStreamPage' instead.
 loadStream
   :: Store
   -> StreamIdentity
   -> Maybe Word64
   -> IO (Either EngineError [StoredEvent])
-loadStream store stream after =
+loadStream store stream = walkPages store stream id
+
+
+-- | Reads a single engine page of a stream, starting after the given sequence.
+loadStreamPage
+  :: Store
+  -> StreamIdentity
+  -> Maybe Word64
+  -> IO (Either EngineError [StoredEvent])
+loadStreamPage store stream after =
   withOpenStore store $ \handle ->
     withInputBuffer (encodeLoadStream stream after) $ \request -> do
       response <- callWithOutput (esLoadStream handle request)
@@ -138,31 +195,21 @@ commit store stream expected events =
       (callWithoutOutput . esCommit handle)
 
 
-supportedAbiMajor :: Word32
-supportedAbiMajor = 0
-
-
-minimumAbiMinor :: Word32
-minimumAbiMinor = 2
-
-
-checkAbiVersion :: Word32 -> Either EngineError ()
-checkAbiVersion version =
-  if actualMajor == supportedAbiMajor && actualMinor >= minimumAbiMinor
-    then Right ()
-    else
-      Left
-        ( AbiVersionMismatch
-            ( AbiVersionDetail
-                supportedAbiMajor
-                minimumAbiMinor
-                actualMajor
-                actualMinor
-            )
-        )
-  where
-    actualMajor = version `shiftR` 16
-    actualMinor = version .&. 0xffff
+walkPages
+  :: Store
+  -> StreamIdentity
+  -> ([StoredEvent] -> [StoredEvent])
+  -> Maybe Word64
+  -> IO (Either EngineError [StoredEvent])
+walkPages store stream loaded after = do
+  page <- loadStreamPage store stream after
+  case page of
+    Left engineError -> pure (Left engineError)
+    Right events -> case NonEmpty.nonEmpty events of
+      Nothing -> pure (Right (loaded []))
+      Just eventPage -> case nextCursor after eventPage of
+        Left engineError -> pure (Left engineError)
+        Right cursor -> walkPages store stream (loaded . (events <>)) (Just cursor)
 
 
 openCompatibleStore :: OpenOptions -> IO (Either EngineError Store)
@@ -194,6 +241,13 @@ closeStoreCell cell = do
   pure ()
 
 
+-- | Hands the owner cell to an engine call for as long as the call runs.
+--
+-- The gate serializes the calls made on one store, so at most one of them is
+-- in flight at a time. Holding the cell through 'withForeignPtr' keeps the
+-- finalizer that frees it from running underneath the call, and the null
+-- check answers a closed store without a foreign call; the engine validates
+-- the cell against its own registry regardless.
 withOpenStore
   :: Store
   -> (Ptr (Ptr EsStore) -> IO (Either EngineError value))
@@ -207,9 +261,16 @@ withOpenStore (Store owner gate) action =
         else action cell
 
 
+-- | Lends an encoded request to an engine call without copying it.
+--
+-- The ABI borrows the request: every entry point decodes out of the buffer
+-- before it returns and retains nothing, so the bytes only have to outlive the
+-- call, which the continuation guarantees. Copying them -- as
+-- useAsCStringLen does, to append a terminator the engine never reads --
+-- would double peak memory on a commit carrying megabytes of payload.
 withInputBuffer :: ByteString -> (Ptr EsBuf -> IO value) -> IO value
 withInputBuffer bytes action =
-  ByteString.useAsCStringLen bytes $ \(pointer, length) ->
+  unsafeUseAsCStringLen bytes $ \(pointer, length) ->
     alloca $ \buffer -> do
       poke buffer (EsBuf (castPtr pointer) (fromIntegral length))
       action buffer
@@ -255,11 +316,14 @@ readOwnedBuffer buffer = do
     then
       if length == 0
         then pure (Right ByteString.empty)
-        else pure (Left (BindingProtocolError "null output buffer with nonzero length"))
+        else
+          pure
+            (Left (BindingProtocolError (NullOutputBufferWithLength length)))
     else
       if length > fromIntegral (maxBound :: Int)
         then
-          pure (Left (BindingProtocolError "engine output exceeds Haskell size limits"))
+          pure
+            (Left (BindingProtocolError (OutputExceedsPlatformSize length)))
         else
           Right
             <$> ByteString.packCStringLen
@@ -272,30 +336,24 @@ readEngineError status buffer = do
   pure case bytes of
     Left protocolError -> protocolError
     Right encoded ->
-      case decodeEngineError (fromIntegral status) encoded of
+      case decodeEngineError code encoded of
         Right engineError -> engineError
         Left cause ->
           BindingProtocolError
-            ( "cannot decode engine error for status "
-                <> Text.pack (show status)
-                <> ": "
-                <> Text.pack cause
+            ( UndecodableEngineError
+                (EngineErrorDecodeDetail code (Text.pack cause))
             )
+  where
+    code :: Word32
+    code = fromIntegral status
 
 
 decodeResponse
   :: (ByteString -> Either String value)
   -> ByteString
   -> Either EngineError value
 decodeResponse decoder =
-  first (BindingProtocolError . Text.pack) . decoder
-
-
-statusWithoutDetail :: CInt -> Either EngineError ()
-statusWithoutDetail 0 = Right ()
-statusWithoutDetail 100 = Left EnginePanic
-statusWithoutDetail status =
-  Left (UnknownEngineError (fromIntegral status))
+  first (BindingProtocolError . UndecodableResponse . Text.pack) . decoder
 
 
 emptyBuffer :: EsBuf