Safe Haskell	None
Language	GHC2021

Ecluse.Core.Osv.Stream

Contents

Ingest bounds and drop accounting
What one attempt learned about its source

Description

Streaming ingest of the osv.dev export archive Pilot compiles osv.db from.

The feed aggregates many upstream databases, so the bounds here are per entry: a drop is counted in IngestStats and the rest of the archive keeps flowing. ilMaxAdvisoryBytes applies before the bytes are retained and before the JSON decodes, and the refused entry drains to its boundary so the entries after it stay aligned. The aggregate verdict is systemicDrop, which the compiler reads once the stream completes.

Synopsis

streamOsvUrl :: forall (m :: Type -> Type) i. (MonadResource m, MonadThrow m, KatipContext m) => Maybe TracerProvider -> OsvIngest -> String -> ConduitT i ExtractedOsv m ()
parseOsvStream :: forall (m :: Type -> Type). (MonadResource m, MonadThrow m, KatipContext m) => Maybe TracerProvider -> OsvIngest -> ConduitT ByteString ExtractedOsv m ()
newtype IngestLimits = IngestLimits {
- ilMaxAdvisoryBytes :: Int
}
defaultIngestLimits :: IngestLimits
data IngestStats = IngestStats {
- statAccepted :: !Int
- statDroppedOversize :: !Int
- statDroppedMalformed :: !Int
- statUnorderable :: !Int
- statUnusableModified :: !Int
}
data OsvIngest
newOsvIngest :: MonadIO m => IngestLimits -> OsvEcosystem -> EpssScores -> UTCTime -> m OsvIngest
readIngestStats :: MonadIO m => OsvIngest -> m IngestStats
resetIngestStats :: MonadIO m => OsvIngest -> m ()
systemicDrop :: IngestStats -> Bool
newtype PilotIngestAborted = PilotIngestAborted IngestStats
data OsvAttempt = OsvAttempt {
- oaLastModified :: Maybe UTCTime
- oaNewestModified :: Maybe UTCTime
}
readOsvAttempt :: MonadIO m => OsvIngest -> m OsvAttempt
resetOsvAttempt :: MonadIO m => OsvIngest -> m ()

Documentation

streamOsvUrl :: forall (m :: Type -> Type) i. (MonadResource m, MonadThrow m, KatipContext m) => Maybe TracerProvider -> OsvIngest -> String -> ConduitT i ExtractedOsv m () Source #

Fetch the OSV zip and stream its contents, bounded by ingest.

parseOsvStream :: forall (m :: Type -> Type). (MonadResource m, MonadThrow m, KatipContext m) => Maybe TracerProvider -> OsvIngest -> ConduitT ByteString ExtractedOsv m () Source #

Parse the zip stream and emit ExtractedOsv, bounded by ingest.

Ingest bounds and drop accounting

newtype IngestLimits Source #

The per-advisory byte bound one ingest pass holds every zip entry to.

Constructors

IngestLimits
Fields ilMaxAdvisoryBytes :: Int Largest decompressed advisory JSON, in bytes, the ingest accepts from one zip entry. It drops a larger one. Bounds memory and, transitively, decode cost.

Instances

Instances details

Show IngestLimits Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods showsPrec :: Int -> IngestLimits -> ShowS # show :: IngestLimits -> String # showList :: [IngestLimits] -> ShowS #
Eq IngestLimits Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods (==) :: IngestLimits -> IngestLimits -> Bool # (/=) :: IngestLimits -> IngestLimits -> Bool #

defaultIngestLimits :: IngestLimits Source #

An 8 MiB per-advisory ceiling.

data IngestStats Source #

The running tally of one ingest pass. Pilot reads it once the stream completes to decide whether the artifact is trustworthy enough to publish (systemicDrop).

Constructors

IngestStats

Fields

statAccepted :: !Int
Advisory entries that decoded successfully.
statDroppedOversize :: !Int
Entries dropped for breaching ilMaxAdvisoryBytes.
statDroppedMalformed :: !Int
Entries dropped because their JSON did not decode.
statUnorderable :: !Int
Rows kept with a bound the grammar cannot parse (orderableBounds). Counted in rows, so it stays out of systemicDrop.
statUnusableModified :: !Int
Records whose modified no grammar reads, or which date it after the run's clock. Their rows are kept and only the date is ignored, so this counts records, not drops.

Instances

Instances details

Show IngestStats Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods showsPrec :: Int -> IngestStats -> ShowS # show :: IngestStats -> String # showList :: [IngestStats] -> ShowS #
Eq IngestStats Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods (==) :: IngestStats -> IngestStats -> Bool # (/=) :: IngestStats -> IngestStats -> Bool #

data OsvIngest Source #

The context one ingest pass threads through the stream.

newOsvIngest :: MonadIO m => IngestLimits -> OsvEcosystem -> EpssScores -> UTCTime -> m OsvIngest Source #

A fresh ingest context with the given bounds, feed, EPSS table and clock, and a zeroed tally.

readIngestStats :: MonadIO m => OsvIngest -> m IngestStats Source #

Read the current drop tally.

resetIngestStats :: MonadIO m => OsvIngest -> m () Source #

Zero the tally. The compiler re-streams from a clean slate on each retry attempt and zeroes the tally alongside it, so the tally reflects only the final attempt.

systemicDrop :: IngestStats -> Bool Source #

Whether the drop tally requires Pilot to refuse publication.

newtype PilotIngestAborted Source #

Raised when systemic drops or zero relevant output prevent publication. The tally records the rejected pass, without replacing a consumer's last-good artifact.

Constructors

PilotIngestAborted IngestStats

Instances

Instances details

Exception PilotIngestAborted Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods toException :: PilotIngestAborted -> SomeException # fromException :: SomeException -> Maybe PilotIngestAborted # displayException :: PilotIngestAborted -> String # backtraceDesired :: PilotIngestAborted -> Bool #
Show PilotIngestAborted Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods showsPrec :: Int -> PilotIngestAborted -> ShowS # show :: PilotIngestAborted -> String # showList :: [PilotIngestAborted] -> ShowS #

What one attempt learned about its source

data OsvAttempt Source #

What one ingest attempt learned about its source beside the rows. A retry replaces it whole, so it always describes the attempt that produced the tally beside it.

Constructors

OsvAttempt
Fields oaLastModified :: Maybe UTCTime The `Last-Modified` the export answered this attempt with. oaNewestModified :: Maybe UTCTime The newest `modified` across the records this attempt read.

Instances

Instances details

Show OsvAttempt Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods showsPrec :: Int -> OsvAttempt -> ShowS # show :: OsvAttempt -> String # showList :: [OsvAttempt] -> ShowS #
Eq OsvAttempt Source #
Instance details Defined in Ecluse.Core.Osv.Stream Methods (==) :: OsvAttempt -> OsvAttempt -> Bool # (/=) :: OsvAttempt -> OsvAttempt -> Bool #

readOsvAttempt :: MonadIO m => OsvIngest -> m OsvAttempt Source #

Read what the current attempt learned about its source.

resetOsvAttempt :: MonadIO m => OsvIngest -> m () Source #

Forget the source metadata, alongside resetIngestStats. A retry re-reads the export, so last attempt's response header and record dates must not survive into this one's artifact.