pipeline

Overview ¶

Package pipeline implements reverse expansion for authorization queries.

Reverse expansion answers the question: "What objects can a user access?" Given a target type/relation and a user, the pipeline traverses the authorization model graph to find all objects where the user has the specified relation.

Architecture ¶

The pipeline constructs a network of workers, one per node in the authorization graph. Workers communicate through message passing, processing tuples and propagating results toward the query origin.

Builder.Build(ctx, graph, spec)
       │
       ▼
  DFS walk ─── builds worker graph from authorization model
       │
       ▼
  ┌─────────┐     ┌─────────┐     ┌─────────┐
  │ Worker  │◄────│ Worker  │◄────│ Worker  │
  │ (root)  │     │         │     │ (leaf)  │
  └─────────┘     └─────────┘     └─────────┘
       │
       ▼
  Pipeline.Recv() ─── results streamed to caller

Trade-offs ¶

Message-Passing Concurrency:

• Workers communicate via bounded buffers, avoiding shared memory
• Natural backpressure prevents memory exhaustion
• Graph structure determines parallelism

Streaming Results:

• Memory usage remains constant regardless of result set size
• Enables early termination when caller calls Close

Cycle Handling:

• Cycles require input deduplication to prevent infinite loops
• Coordinated shutdown ensures all in-flight messages are processed

Usage ¶

builder, err := pipeline.NewBuilder(store,
    pipeline.WithNumProcs(4),
    pipeline.WithChunkSize(100),
)
if err != nil {
    return err
}

p, err := builder.Build(ctx, graph, pipeline.Spec{
    ObjectType:     "document",
    ObjectRelation: "viewer",
    SubjectType:    "user",
    SubjectID:      "alice",
})
if err != nil {
    return err
}
defer p.Close()

for {
    id, ok := p.Recv(ctx)
    if !ok {
        break
    }
    // id is an object ID like "document:readme"
}

if err := p.Err(); err != nil {
    // handle error (e.g. storage failure, context cancellation)
    return err
}