use super::{GrantStreamConfig, StreamError, StreamState};
use crate::auth_membership::{MembershipChange, MembershipGate, MembershipListener};
use crate::auth_revocation::{
Applied, Digest32, GrantFeedChange, GrantFeedEvent, GrantId, GrantLogId, Revision,
RevocationAuthority,
};
use lys_identity::grants::change_stream::{ChangeFrame, ChangeKind, ChangesPage, event_bytes};
use lys_identity::grants::{GrantChange, verify_grant_event};
use std::sync::Arc;
pub(super) struct Engine {
pub state: Arc<StreamState>,
pub authority: Arc<RevocationAuthority>,
pub config: GrantStreamConfig,
pub service_key: [u8; 32],
pub gate: Option<Arc<MembershipGate>>,
pub supervisor: Option<crate::server::ConnectionSupervisor>,
needs_resume: bool,
root_owed: bool,
placement_barrier: Option<u64>,
}
impl Engine {
pub const fn new(
state: Arc<StreamState>,
authority: Arc<RevocationAuthority>,
config: GrantStreamConfig,
service_key: [u8; 32],
gate: Option<Arc<MembershipGate>>,
supervisor: Option<crate::server::ConnectionSupervisor>,
) -> Self {
Self {
state,
authority,
config,
service_key,
gate,
supervisor,
needs_resume: true,
root_owed: false,
placement_barrier: None,
}
}
fn log(&self) -> Result<GrantLogId, StreamError> {
GrantLogId::new(&self.state.log.identity, self.state.log.epoch)
.map_err(|_| StreamError::named("revocation_log_reset"))
}
fn apply(&self, event: &GrantFeedEvent) -> Result<Applied, StreamError> {
let outcome = match &self.supervisor {
Some(supervisor) => supervisor
.apply_grant_event(event)
.map_err(|_| StreamError::named("grant_stream_authority_missing"))?,
None => self.authority.apply(event),
};
match outcome.applied {
Applied::Gap => {
let through = match event {
GrantFeedEvent::Change { revision, .. } => {
Some(revision.get().saturating_sub(1))
}
_ => None,
};
Err(self.gap(through)?)
}
Applied::Terminal(_) | Applied::Refused(_) => {
Err(StreamError::named("grant_stream_evidence_invalid"))
}
applied => Ok(applied),
}
}
fn gap(&self, through: Option<u64>) -> Result<StreamError, StreamError> {
self.state.update(|inner| {
inner.diagnostics.readiness = "gap";
inner.diagnostics.last_error = Some("revocation_stream_gap");
inner.diagnostics.missing = through.and_then(|through| {
inner
.diagnostics
.cursor
.and_then(|cursor| through.checked_sub(cursor))
});
Ok(())
})?;
Ok(StreamError::named("revocation_stream_gap"))
}
pub fn lost(&mut self) -> Result<(), StreamError> {
self.needs_resume = true;
self.apply(&GrantFeedEvent::Lost)?;
self.state.unavailable("revocation_stream_unavailable")
}
pub fn terminal(&self, reason: &'static str) -> Result<(), StreamError> {
use crate::auth_revocation::Readiness;
let status = self
.authority
.diagnostics()
.map_err(|_| StreamError::named("grant_stream_authority_poisoned"))?;
if matches!(
status.readiness,
Readiness::Gap | Readiness::Reset | Readiness::Terminal
) {
return Ok(());
}
let event = GrantFeedEvent::Invalid { reason };
let applied = match &self.supervisor {
Some(supervisor) => {
supervisor
.apply_grant_event(&event)
.map_err(|_| StreamError::named("grant_stream_authority_missing"))?
.applied
}
None => self.authority.apply(&event).applied,
};
if !matches!(applied, Applied::Terminal(_)) {
return Err(StreamError::named("grant_stream_terminal_fence_failed"));
}
self.state.update(|inner| {
inner.diagnostics.readiness = "terminal";
inner.diagnostics.last_error = Some(reason);
Ok(())
})
}
pub fn page(&mut self, page: ChangesPage) -> Result<(), StreamError> {
self.placement_barrier = None;
if page.frames.is_empty() || page.frames.len() > self.config.max_frames {
return Err(StreamError::named("grant_stream_frame_over_bound"));
}
let explicit_reset = matches!(page.frames.first(), Some(ChangeFrame::Reset { .. }));
if page.log.identity != self.state.log.identity
|| page.log.epoch != self.state.log.epoch
|| explicit_reset
{
if explicit_reset {
if let Some(ChangeFrame::Reset {
from: Some(from), ..
}) = page.frames.first()
&& (from.identity != self.state.log.identity
|| from.epoch != self.state.log.epoch)
{
return Err(StreamError::named("grant_stream_reset_foreign"));
}
let to = GrantLogId::new(&page.log.identity, page.log.epoch)
.map_err(|_| StreamError::named("grant_stream_reset_invalid"))?;
self.apply(&GrantFeedEvent::Reset {
from: self.log()?,
to,
})?;
} else {
self.apply(&GrantFeedEvent::Lost)?;
}
self.state.update(|inner| {
inner.diagnostics.readiness = "reset";
inner.diagnostics.last_error = Some("revocation_log_reset");
Ok(())
})?;
return Err(StreamError::named("revocation_log_reset"));
}
self.apply(&GrantFeedEvent::Lost)?;
self.state.update(|inner| {
inner.diagnostics.readiness = "not_ready";
Ok(())
})?;
let first = self.state.snapshot()?;
let mut positions = 0_u64;
let length = page.frames.len();
for (index, frame) in page.frames.into_iter().enumerate() {
let status = self.state.snapshot()?;
let count = match &frame {
ChangeFrame::Change { .. } => 1,
ChangeFrame::Watermark { revision } => status
.cursor
.map_or(0, |cursor| revision.saturating_sub(cursor)),
ChangeFrame::Placement { placement, .. }
| ChangeFrame::PlacementMark { placement, .. } => status
.placed
.map_or(0, |placed| placement.saturating_sub(placed)),
_ => 0,
};
if first.cursor.is_some() {
positions = positions
.checked_add(count)
.ok_or(StreamError::named("grant_stream_frame_over_bound"))?;
if positions > self.config.page_positions {
return Err(StreamError::named("grant_stream_frame_over_bound"));
}
}
if matches!(
frame,
ChangeFrame::Ready { .. } | ChangeFrame::Unready { .. }
) && index + 1 != length
{
return Err(StreamError::named("grant_stream_barrier_invalid"));
}
self.frame(frame)?;
}
Ok(())
}
fn frame(&mut self, frame: ChangeFrame) -> Result<(), StreamError> {
match frame {
ChangeFrame::Baseline { revision, revoked } => self.baseline(revision, revoked),
ChangeFrame::Change {
revision,
grant,
change,
event,
} => self.change(revision, &grant, change, &event),
ChangeFrame::Watermark { revision } => {
self.needs_resume = true;
self.apply(&GrantFeedEvent::Watermark {
log: self.log()?,
through: Revision::new(revision),
})?;
self.advance(revision)
}
ChangeFrame::Placement {
revision,
placement,
..
} => {
let status = self.state.snapshot()?;
if status.cursor != Some(revision)
|| status.placed.and_then(|placed| placed.checked_add(1)) != Some(placement)
{
return Err(self.gap(None)?);
}
self.needs_resume = true;
self.placed(placement)
}
ChangeFrame::PlacementMark {
revision,
placement,
} => {
let status = self.state.snapshot()?;
if status.cursor.is_none_or(|cursor| revision > cursor)
|| status.placed.is_some_and(|placed| placement < placed)
{
return Err(self.gap(None)?);
}
if status.placed != Some(placement) {
self.needs_resume = true;
}
self.placement_barrier = Some(revision);
self.placed(placement)
}
ChangeFrame::Ready { revision } => self.ready(revision),
ChangeFrame::Unready { .. } => {
self.needs_resume = true;
self.apply(&GrantFeedEvent::Lost)?;
self.state.unavailable("revocation_stream_unavailable")
}
ChangeFrame::Reset { .. } => Err(StreamError::named("grant_stream_reset_invalid")),
}
}
fn baseline(&self, revision: u64, revoked: Vec<String>) -> Result<(), StreamError> {
if self.state.snapshot()?.cursor.is_some() {
return Err(StreamError::named("grant_stream_baseline_unexpected"));
}
if revoked.len() > self.config.max_revoked_grants {
return Err(StreamError::named("grant_stream_revoked_over_bound"));
}
let mut seen = std::collections::BTreeSet::new();
for id in revoked {
let parsed: lys_identity::grants::GrantId = id
.parse()
.map_err(|_| StreamError::named("grant_stream_grant_invalid"))?;
let grant = GrantId::new(&parsed.to_string())
.map_err(|_| StreamError::named("grant_stream_grant_invalid"))?;
if !seen.insert(grant) {
return Err(StreamError::named("grant_stream_baseline_duplicate"));
}
}
let decisions = seen.iter().cloned().map(|grant| (grant, true)).collect();
self.apply(&GrantFeedEvent::Baseline {
log: self.log()?,
through: Revision::new(revision),
decisions,
})?;
self.state.update(|inner| {
inner.revoked = seen;
inner.diagnostics.cursor = Some(revision);
Ok(())
})
}
fn advance(&self, revision: u64) -> Result<(), StreamError> {
self.state.update(|inner| {
if inner
.diagnostics
.cursor
.is_none_or(|cursor| revision < cursor)
{
return Err(StreamError::named("grant_stream_cursor_regressed"));
}
inner.diagnostics.cursor = Some(revision);
Ok(())
})
}
fn placed(&self, position: u64) -> Result<(), StreamError> {
self.state.update(|inner| {
inner.diagnostics.placed = Some(position);
Ok(())
})
}
fn change(
&mut self,
revision: u64,
grant: &str,
kind: ChangeKind,
event: &str,
) -> Result<(), StreamError> {
if event.len() / 2 > self.config.max_event_bytes {
return Err(StreamError::named("grant_stream_event_over_bound"));
}
let bytes =
event_bytes(event).ok_or(StreamError::named("grant_stream_evidence_invalid"))?;
let signed = verify_grant_event(&bytes, &self.service_key)
.map_err(|_| StreamError::named("grant_stream_evidence_invalid"))?;
if signed.event().grant().to_string() != grant {
return Err(StreamError::named("grant_stream_evidence_invalid"));
}
let change = match (kind, signed.event().change()) {
(ChangeKind::Issue, GrantChange::Issue(_)) => GrantFeedChange::Issue(
GrantId::new(grant)
.map_err(|_| StreamError::named("grant_stream_grant_invalid"))?,
),
(ChangeKind::Revoke, GrantChange::Revoke { .. }) => GrantFeedChange::Revoke(
GrantId::new(grant)
.map_err(|_| StreamError::named("grant_stream_grant_invalid"))?,
),
_ => return Err(StreamError::named("grant_stream_evidence_invalid")),
};
if matches!(kind, ChangeKind::Revoke) {
self.state.update(|inner| {
if !inner.revoked.contains(
&GrantId::new(grant)
.map_err(|_| StreamError::named("grant_stream_grant_invalid"))?,
) && inner.revoked.len() >= self.config.max_revoked_grants
{
return Err(StreamError::named("grant_stream_revoked_over_bound"));
}
Ok(())
})?;
}
let applied = self.apply(&GrantFeedEvent::Change {
log: self.log()?,
revision: Revision::new(revision),
change,
evidence: Digest32(*blake3::hash(&bytes).as_bytes()),
})?;
if applied == Applied::Duplicate {
return Ok(());
}
self.advance(revision)?;
match signed.event().change() {
GrantChange::Issue(granted) if !self.needs_resume => {
let change = MembershipChange {
log: self.state.log.clone(),
revision,
subject: None,
resource: lys_pass::rights::Resource {
kind: granted.resource().kind().into(),
id: granted.resource().id().into(),
},
};
if let Some(gate) = &self.gate {
let report = gate
.changed(&change)
.map_err(|_| StreamError::named("grant_stream_membership_failed"))?;
if report.failed > 0 {
return Err(StreamError::named("grant_stream_membership_failed"));
}
}
self.needs_resume = true;
}
GrantChange::Revoke { .. } => {
self.needs_resume = true;
let grant = GrantId::new(grant)
.map_err(|_| StreamError::named("grant_stream_grant_invalid"))?;
self.state.update(|inner| {
inner.revoked.insert(grant);
Ok(())
})?;
}
_ => {}
}
Ok(())
}
pub const fn take_root_owed(&mut self) -> bool {
let owed = self.root_owed;
self.root_owed = false;
owed
}
fn ready(&mut self, revision: u64) -> Result<(), StreamError> {
let status = self.state.snapshot()?;
if status.cursor != Some(revision) {
return Err(self.gap(Some(revision))?);
}
if status.placed.is_none() || self.placement_barrier != Some(revision) {
return Err(StreamError::named("grant_stream_placement_barrier_missing"));
}
let generation = if self.needs_resume {
status
.generation
.checked_add(1)
.ok_or(StreamError::named("grant_stream_generation_exhausted"))?
} else {
status.generation
};
if self.needs_resume
&& let Some(gate) = &self.gate
{
let report = gate
.resumed(generation, revision)
.map_err(|_| StreamError::named("grant_stream_membership_failed"))?;
if !report.settled {
return Err(StreamError::named("grant_stream_membership_failed"));
}
}
self.apply(&GrantFeedEvent::Ready {
log: self.log()?,
through: Revision::new(revision),
})?;
self.root_owed |= self.needs_resume;
self.needs_resume = false;
self.state.update(|inner| {
inner.diagnostics.readiness = "ready";
inner.diagnostics.generation = generation;
inner.diagnostics.last_error = None;
inner.diagnostics.missing = None;
Ok(())
})
}
#[cfg(test)]
pub fn for_tests(state: Arc<StreamState>) -> Result<Self, StreamError> {
use crate::auth_revocation::{RevocationPolicy, UnpublishedGrantBinding};
let config = super::tests::configuration()?;
let policy = RevocationPolicy {
log: GrantLogId::new(&state.log.identity, state.log.epoch)
.map_err(|_| StreamError::named("grant_stream_log_invalid"))?,
audience: "liminal".into(),
issuer: Digest32([0; 32]),
};
let authority = Arc::new(RevocationAuthority::new(
policy,
Arc::new(UnpublishedGrantBinding),
state.clone(),
));
Ok(Self::new(state, authority, config, [0; 32], None, None))
}
}
impl Drop for Engine {
fn drop(&mut self) {
let unfenced = match self.authority.diagnostics() {
Ok(status) => status.readiness == crate::auth_revocation::Readiness::Ready,
Err(_) => true,
};
if unfenced {
if let Err(error) = self
.state
.unavailable("grant_stream_worker_ended_without_fence")
{
tracing::error!(cause = error.name(), "grant stream membership fence failed");
}
if let Err(error) = self.terminal("grant_stream_worker_ended_without_fence") {
tracing::error!(cause = error.name(), "grant stream admission fence failed");
}
}
}
}