diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2025-2026 hypermemetic + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/plexus-protocol.cabal b/plexus-protocol.cabal --- a/plexus-protocol.cabal +++ b/plexus-protocol.cabal @@ -1,11 +1,19 @@ cabal-version: 3.0 name: plexus-protocol -version: 0.5.2.0 +version: 0.5.5.0 synopsis: Plexus RPC protocol types and client +description: Haskell types and a client for Plexus RPC, a streaming JSON-RPC 2.0 + protocol: stream items and provenance, recursive plugin and method + schemas, authentication, and a WebSocket transport. Shared by Synapse + and other Plexus RPC clients. license: MIT -author: -maintainer: +license-file: LICENSE +author: hypermemetic +maintainer: hypermemetic@proton.me +copyright: (c) 2025-2026 hypermemetic +category: Network build-type: Simple +tested-with: GHC == 9.8.4, GHC == 9.12.4 library exposed-modules: @@ -19,7 +27,7 @@ Plexus.Schema.Cache Plexus.Schema.Recursive build-depends: - base ^>=4.17 || ^>=4.18 || ^>=4.19, + base ^>=4.17 || ^>=4.18 || ^>=4.19 || ^>=4.21, aeson >= 2.0, aeson-casing >= 0.2, text >= 2.0, @@ -49,3 +57,18 @@ GeneralizedNewtypeDeriving LambdaCase RecordWildCards + +-- NAMING-9: the naming gate (scripts/naming-gate.sh) in the suite. The +-- spec counts the gate's lines and shows the first; it never diffs them. +-- Its heap is capped (-M6G; +RTS -M... -RTS overrides it), so no regression +-- of that can take the machine (2026-10-29: an hspec diff reached 7.3 GB). +test-suite naming-gate + type: exitcode-stdio-1.0 + main-is: NamingGateSpec.hs + build-depends: + base >= 4.17 && < 5, + process >= 1.6 && < 1.7, + hspec >= 2.10 && < 2.12 + hs-source-dirs: test + default-language: GHC2021 + ghc-options: -threaded -rtsopts "-with-rtsopts=-M6G" diff --git a/src/Plexus/Auth.hs b/src/Plexus/Auth.hs --- a/src/Plexus/Auth.hs +++ b/src/Plexus/Auth.hs @@ -1,7 +1,7 @@ --- | AUTH-4 / AUTH-2 / AUTH-5: what a client parses of a hub's auth. +-- | AUTH-4 / AUTH-2 / AUTH-5: what a client parses of a service's auth. -- --- Every field is optional on the wire: a hub without a gate (every hub --- before AUTH, and any hub that leaves login off) answers none of it, and +-- Every field is optional on the wire: a service without a gate (every service +-- before AUTH, and any service that leaves login off) answers none of it, and -- everything here parses its absence as 'Nothing' or empty. -- -- * '_info' gains @view@ (@anonymous@ | @caller@), the view's @hash@, and, @@ -36,7 +36,7 @@ -- The view -- ============================================================================ --- | Which part of a hub the caller sees: an anonymous caller only how to +-- | Which part of a service the caller sees: an anonymous caller only how to -- sign in; a signed-in caller what it can use. data View = Anonymous | Caller deriving stock (Show, Eq, Ord, Generic) @@ -46,7 +46,7 @@ viewText Caller = "caller" -- | @"anonymous"@ / @"caller"@, or an object naming it as @kind@ (a --- tolerance: a hub may carry the view's hash beside it). +-- tolerance: a service may carry the view's hash beside it). instance FromJSON View where parseJSON = \case String s -> fromName s @@ -82,7 +82,7 @@ -- | What @_info@ answers. Only @backend@ is required. data BackendInfo = BackendInfo { infoBackend :: Text - , infoView :: Maybe View -- ^ Nothing: a hub with no gate (today's) + , infoView :: Maybe View -- ^ Nothing: a service with no gate (today's) , infoViewHash :: Maybe Text -- ^ the view's own hash (@view_hash@, else @hash@, else the view object's @hash@) , infoAuth :: Maybe AuthInfo } @@ -132,7 +132,7 @@ , refusalScope :: Maybe Text -- ^ where it was asked: @IDENTITY@ , refusalRoles :: [Text] -- ^ the roles that list the grant , refusalHeldOn :: [Text] -- ^ where the caller holds it - , refusalCeiling :: [Text] -- ^ the session's ceiling, when the hub says it + , refusalCeiling :: [Text] -- ^ the session's ceiling, when the service says it , refusalMessage :: Text } deriving stock (Show, Eq, Generic) diff --git a/src/Plexus/Client.hs b/src/Plexus/Client.hs --- a/src/Plexus/Client.hs +++ b/src/Plexus/Client.hs @@ -22,6 +22,8 @@ -- * Core RPC primitive , substrateRpc , substrateRpcWith + , substrateRpcSub + , unsubscribe -- * Configuration , SubstrateConfig(..) @@ -350,6 +352,40 @@ go (Just now) | otherwise -> go Nothing +-- | SYNAPSE-28: end a subscription the caller no longer reads — the +-- server's @unsubMethod@ (@.call_unsub@) with its id, as the +-- server gave it (a number stays a number). The answer is not waited for; +-- nothing more is delivered to the call's queue. +unsubscribe :: SubstrateConnection -> Text -> SubscriptionId -> IO () +unsubscribe conn unsubMethod subId = do + atomically $ modifyTVar' (scSubscriptions conn) (Map.delete subId) + gone <- connectionLost conn + case gone of + Just _ -> pure () + Nothing -> do + rid <- atomicModifyIORef' (scNextId conn) $ \n -> (n + 1, RequestId n) + -- the answer is taken and dropped by the reader (a pending request + -- no one waits on) + queue <- newTQueueIO + respVar <- newEmptyTMVarIO + atomically $ modifyTVar' (scPendingReqs conn) (Map.insert rid (PendingRequest queue respVar)) + let raw = unSubscriptionId subId + idValue = case decodeStrictText raw of + Just n@(Number _) -> n + _ -> String raw + req = object + [ "jsonrpc" .= ("2.0" :: Text), "id" .= rid + , "method" .= unsubMethod, "params" .= [idValue] ] + WS.sendTextData (scConnection conn) (encode req) + `catch` \(_ :: SomeException) -> pure () + -- forget it shortly; the reader has filled respVar by then, or never will + void $ forkIO $ do + threadDelay 30000000 + atomically $ modifyTVar' (scPendingReqs conn) (Map.delete rid) + where + decodeStrictText :: Text -> Maybe Value + decodeStrictText = decode . LBS.fromStrict . BS8.pack . T.unpack + -- | Disconnect from substrate disconnect :: SubstrateConnection -> IO () disconnect SubstrateConnection{..} = do @@ -452,7 +488,19 @@ -> Text -> Value -> Stream (Of PlexusStreamItem) IO () -substrateRpcWith conn onSubscribed method params = do +substrateRpcWith conn onSubscribed = substrateRpcSub conn (const onSubscribed) + +-- | SYNAPSE-28: 'substrateRpcWith', telling @onSubscribed@ the +-- subscription's id — so a caller that shares the connection (a holder +-- serving many calls on one socket) can end a call it abandons with +-- 'unsubscribe'. +substrateRpcSub + :: SubstrateConnection + -> (SubscriptionId -> IO ()) + -> Text + -> Value + -> Stream (Of PlexusStreamItem) IO () +substrateRpcSub conn onSubscribed method params = do -- SYNAPSE-20: a connection already gone (it died idle in a pool) takes -- no request; nothing was sent, so the caller may always try again gone <- liftIO $ connectionLost conn @@ -507,7 +555,7 @@ pure () Success subId -> do - liftIO onSubscribed + liftIO (onSubscribed subId) -- Stream items until done, cleanup when finished streamItems queue subId <* liftIO (cleanup subId) where diff --git a/src/Plexus/Schema/Recursive.hs b/src/Plexus/Schema/Recursive.hs --- a/src/Plexus/Schema/Recursive.hs +++ b/src/Plexus/Schema/Recursive.hs @@ -49,7 +49,8 @@ , xPlexusValues -- * Queries - , isHubActivation + , isParentActivation + , isHubActivation -- naming-alias: deprecated, removed by NAMING-18 , isLeafActivation , pluginMethods , pluginChildren @@ -228,6 +229,7 @@ , methodPublic :: Bool -- ^ AUTH-3: callable anonymously (the auth activation's door) , methodGroup :: Maybe Text -- ^ SYNAPSE-13: the method's group in a listing (OpenAPI's tag) , methodExamples :: [MethodExample] -- ^ SYNAPSE-13: worked calls, each its params and a note + , methodInput :: Maybe Value -- ^ SYNAPSE-CC-8 / BIDIR-6: the JSON Schema of what the client sends into a call after it starts (@input@); absent on every server until BIDIR-6 } deriving stock (Show, Eq, Generic) @@ -251,9 +253,10 @@ <*> o .:? "target" <*> o .:? "held_on" <*> o .:? "public" .!= False - -- SYNAPSE-13: absent on an older hub; an unreadable one is ignored + -- SYNAPSE-13: absent on an older service; an unreadable one is ignored <*> (o .:? "group" <|> pure Nothing) <*> (o .:? "examples" .!= [] <|> pure []) + <*> o .:? "input" instance ToJSON MethodSchema where toJSON MethodSchema{..} = object $ @@ -277,6 +280,7 @@ , if methodPublic then Just ("public" .= True) else Nothing , ("group" .=) <$> methodGroup , if null methodExamples then Nothing else Just ("examples" .= methodExamples) + , ("input" .=) <$> methodInput ] -- | SYNAPSE-13: one worked call of a method — the params it sends and a @@ -320,7 +324,7 @@ , psLongDescription :: Maybe Text -- ^ Extended description (no word limit) , psHash :: PluginHash , psMethods :: [MethodSchema] - , psChildren :: Maybe [ChildSummary] -- ^ Nothing = leaf, Just = hub activation + , psChildren :: Maybe [ChildSummary] -- ^ Nothing = leaf, Just = parent activation , psDeprecation :: Maybe DeprecationInfo -- ^ Activation-level deprecation (IR-5) , psRoles :: Maybe (Map Text [Text]) -- ^ AUTH-3: the activation's default roles, each its grants , psView :: Maybe View -- ^ AUTH-4: which view this schema answers from @@ -382,13 +386,18 @@ hasAuthColumns :: PluginSchema -> Bool hasAuthColumns ps = any (\m -> maybe False (const True) (methodGrant m) || maybe False (const True) (methodHeldOn m)) (psMethods ps) --- | Is this a hub activation (has children)? -isHubActivation :: PluginSchema -> Bool -isHubActivation = maybe False (not . null) . psChildren +-- | Is this a parent activation (has children)? +isParentActivation :: PluginSchema -> Bool +isParentActivation = maybe False (not . null) . psChildren -- | Is this a leaf activation (no children)? isLeafActivation :: PluginSchema -> Bool -isLeafActivation = not . isHubActivation +isLeafActivation = not . isParentActivation + +-- | The old name of 'isParentActivation', kept for one release. -- naming-alias: removed by NAMING-18 +{-# DEPRECATED isHubActivation "renamed isParentActivation — NAMING" #-} -- naming-alias +isHubActivation :: PluginSchema -> Bool -- naming-alias +isHubActivation = isParentActivation -- naming-alias -- | Get methods (alias for psMethods) pluginMethods :: PluginSchema -> [MethodSchema] diff --git a/src/Plexus/Transport.hs b/src/Plexus/Transport.hs --- a/src/Plexus/Transport.hs +++ b/src/Plexus/Transport.hs @@ -24,11 +24,16 @@ -- * Method Invocation (collected) , invokeMethod , invokeRaw + + -- * Routing a call elsewhere (SYNAPSE-28) + , CallRoute + , setCallRouter ) where import Control.Exception (SomeException, IOException, catch, fromException) import qualified Control.Exception as E import Control.Concurrent.MVar (MVar, newMVar, modifyMVar) +import Data.IORef (IORef, newIORef, readIORef, writeIORef, modifyIORef') import Data.Aeson hiding (Error) import Data.Map.Strict (Map) import qualified Data.Map.Strict as Map @@ -58,6 +63,24 @@ pool <- createConnectionPool cfg defaultPlexusPoolConfig pure (Map.insert cfg pool pools, pool) +-- | SYNAPSE-28: one call, made some other way than a pooled connection of +-- this process — @onSubscribed method params onItem@ — e.g. through a +-- connection held between runs, or one connection a whole process shares. +-- It reports a lost connection as 'Plexus.Client.connect' and +-- 'substrateRpcWith' do ('LostConnection', a failed dial as an error +-- call), so everything above it reads its failures as before. +type CallRoute = IO () -> Text -> Value -> (PlexusStreamItem -> IO ()) -> IO () + +{-# NOINLINE routerRef #-} +routerRef :: IORef (SubstrateConfig -> Maybe CallRoute) +routerRef = unsafePerformIO (newIORef (const Nothing)) + +-- | SYNAPSE-28: route the calls whose config the router answers 'Just' +-- for; the rest go through this process's pools, as before. Set once, at +-- start (a program that never sets it is unchanged). +setCallRouter :: (SubstrateConfig -> Maybe CallRoute) -> IO () +setCallRouter = writeIORef routerRef + -- | Low-level RPC call with default localhost config rpcCall :: Text -> Text -> Value -> IO (Either TransportError [PlexusStreamItem]) rpcCall backend = rpcCallWith (defaultConfig backend) @@ -107,10 +130,17 @@ on f g x y = f (g x) (g y) doCallInner :: SubstrateConfig -> Text -> Value -> IO [PlexusStreamItem] -doCallInner cfg method params = freshIfUnsent $ do - pool <- getOrCreatePool cfg - withPooledConnection pool $ \conn -> - S.toList_ $ substrateRpc conn method params +doCallInner cfg method params = do + router <- readIORef routerRef + case router cfg of + Just route -> do + acc <- newIORef [] + route (pure ()) method params (\i -> modifyIORef' acc (i :)) + reverse <$> readIORef acc + Nothing -> freshIfUnsent $ do + pool <- getOrCreatePool cfg + withPooledConnection pool $ \conn -> + S.toList_ $ substrateRpc conn method params -- | SYNAPSE-20: a pooled connection that died while idle fails the call -- before anything is sent (and the pool drops it); the call goes once more, @@ -131,10 +161,14 @@ doCallStreaming cfg = doCallStreamingWith cfg (pure ()) doCallStreamingWith :: SubstrateConfig -> IO () -> Text -> Value -> (PlexusStreamItem -> IO ()) -> IO () -doCallStreamingWith cfg onSubscribed method params onItem = freshIfUnsent $ do - pool <- getOrCreatePool cfg - withPooledConnection pool $ \conn -> - S.mapM_ onItem $ substrateRpcWith conn onSubscribed method params +doCallStreamingWith cfg onSubscribed method params onItem = do + router <- readIORef routerRef + case router cfg of + Just route -> route onSubscribed method params onItem + Nothing -> freshIfUnsent $ do + pool <- getOrCreatePool cfg + withPooledConnection pool $ \conn -> + S.mapM_ onItem $ substrateRpcWith conn onSubscribed method params -- | Streaming method invocation invokeMethodStreaming :: SubstrateConfig -> [Text] -> Text -> Value -> (PlexusStreamItem -> IO ()) -> IO (Either TransportError ()) diff --git a/src/Plexus/Types.hs b/src/Plexus/Types.hs --- a/src/Plexus/Types.hs +++ b/src/Plexus/Types.hs @@ -493,7 +493,7 @@ , itemRecoverable :: Bool , itemErrorRefusal :: Maybe Refusal -- ^ AUTH-2: @error_kind: "forbidden"@ and what it names; Nothing - -- for every other error (and on every hub without a gate) + -- for every other error (and on every service without a gate) } | StreamRequest { itemPlexusHash :: Text @@ -628,7 +628,7 @@ , "recoverable" .= rec ] <> refusalFields where - -- the refusal's own fields, beside the message (as the hub sent them) + -- the refusal's own fields, beside the message (as the service sent them) refusalFields = case toJSON <$> refusal of Just (Object ro) -> [ (k, v) | (k, v) <- KM.toList ro, k /= "message" ] _ -> [] diff --git a/test/NamingGateSpec.hs b/test/NamingGateSpec.hs new file mode 100644 --- /dev/null +++ b/test/NamingGateSpec.hs @@ -0,0 +1,42 @@ +-- | NAMING-9: a Plexus backend is a service, never the retired word (owner, +-- 2026-10-29; ~/AGENTS.md "Naming"). +-- The suite runs the repository's gate, scripts/naming-gate.sh, so +-- @cabal test@ — and CI — fails the day the retired word is written again. +-- +-- A failure says how many lines the gate found and shows the first of them. +-- The gate's output is counted, never compared whole with an expected value: +-- on 2026-10-29 hspec's diff of an 875-line output grew to 7.3 GB and the +-- kernel's OOM killer stopped it (and the session around it). The suite's +-- heap is capped too (-with-rtsopts=-M6G in the cabal file). +module Main (main) where + +import Control.Monad (unless) +import GHC.IO.Encoding (setLocaleEncoding) +import System.Exit (ExitCode (..)) +import System.IO (hSetEncoding, stderr, stdout, utf8) +import System.Process (readProcessWithExitCode) +import Test.Hspec + +-- | How many of the gate's lines a failure shows; @bash scripts/naming-gate.sh@ +-- prints every one. +shown :: Int +shown = 20 + +main :: IO () +main = do + -- the gate prints lines of UTF-8 files; read and show them as such + -- whatever the locale (a C locale would refuse the first em dash) + setLocaleEncoding utf8 + mapM_ (`hSetEncoding` utf8) [stdout, stderr] + hspec $ + describe "scripts/naming-gate.sh" $ + it "finds no line that says the retired word (exit 0)" $ do + (code, out, err) <- readProcessWithExitCode "bash" ["scripts/naming-gate.sh"] "" + let found = lines out + n = length found + unless (code == ExitSuccess && n == 0) . expectationFailure . unlines $ + [ "the gate found " <> show n <> " line(s) (" <> show code <> "); the first " + <> show (min n shown) <> ":" ] + <> take shown found + <> [ "... and " <> show (n - shown) <> " more: bash scripts/naming-gate.sh" | n > shown ] + <> take 3 (lines err)