Documentation
¶
Overview ¶
Package subscriptions implements MCP resource subscriptions (resources/subscribe) over GitLab resources.
Delivery is polling-based, in both transports: a watcher re-reads a subscribed URI on an interval and emits notifications/resources/updated only when the content actually changed. There is no webhook path — this server does not run an inbound receiver of any kind, a decision recorded in ADR-0016 (docs/development/adr/adr-0016-no-webhook-ingestion.md), not a gap left for later.
Index ¶
- Constants
- Variables
- func Templates() []string
- func TranslateReadError(err error) error
- type Kind
- type Manager
- func (m *Manager[S]) Close()
- func (m *Manager[S]) Len() int
- func (m *Manager[S]) Renew(uri string) bool
- func (m *Manager[S]) RenewAll(subscriber S) int
- func (m *Manager[S]) Subscribe(ctx context.Context, subscriber S, uri string) error
- func (m *Manager[S]) Unsubscribe(subscriber S, uri string) error
- func (m *Manager[S]) UnsubscribeAll(subscriber S) int
- type Notifier
- type Options
- type Reader
- type Update
Constants ¶
const ( DefaultBaseInterval = 15 * time.Second DefaultMinInterval = 5 * time.Second DefaultLease = 30 * time.Minute DefaultSlowInterval = 10 * time.Minute DefaultMaxLifetime = 24 * time.Hour DefaultMaxWatchers = 10 )
Defaults for Options. The interval numbers come from GitLab's rate limits rather than from taste: a self-managed instance that enables the optional authenticated-API throttle gets 120 requests a minute by default (the throttle itself ships disabled — see ADR-0017), so ten watchers at a five-second interval would consume such a user's entire budget while that same user is making tool calls through it.
Variables ¶
var ( // ErrInaccessible means the resource can no longer be read — it was // deleted, or this token lost access to it. GitLab answers 404 for // both cases so as not to leak existence, and the watcher treats them // the same way: stop watching. Continuing would burn budget on a // resource that can never produce a notification, and would keep // polling with a token whose access was deliberately revoked. ErrInaccessible = errors.New("subscriptions: resource inaccessible") // ErrRateLimited means GitLab refused the read for rate-limiting // reasons. It pauses every watcher on this manager, not just the one // that hit it: the limit is enforced per user, so the others are // about to hit it too. ErrRateLimited = errors.New("subscriptions: rate limited") )
Errors a Reader may return to steer a watcher. Everything else is treated as a transient failure: logged, and retried on the next tick.
var ( // ErrLifetimeExceeded means the watch hit [Options.MaxLifetime]. ErrLifetimeExceeded = errors.New("subscriptions: maximum watch lifetime reached") // ErrEvicted means the watch was stopped to make room for one whose // subscriber is still active. ErrEvicted = errors.New("subscriptions: evicted to make room for an active subscription") )
Reasons reported to Options.OnStop — the ways a watch can end without the client having asked for it.
var ErrClosed = errors.New("subscriptions: manager is closed")
ErrClosed is returned once the manager has been shut down.
var ErrNotSubscribable = errors.New("subscriptions: resource is not subscribable")
ErrNotSubscribable is returned by Manager.Subscribe for a URI outside the whitelist. The SDK does not check that a subscribed URI names a registered resource, so this is the only thing standing between a client and a subscription that can never fire.
var ErrTooManySubscriptions = errors.New("subscriptions: too many active subscriptions")
ErrTooManySubscriptions is returned when a manager is already watching its configured maximum.
Functions ¶
func Templates ¶
func Templates() []string
Template returns the URI template this kind's resource is registered under, or "" for KindUnknown. Templates returns the URI template of every subscribable kind, sorted, for surfaces that advertise what can be watched (the gitlab://tools manifest). Deriving the list here rather than copying it means the advertisement can never drift from the whitelist that enforces it.
func TranslateReadError ¶
TranslateReadError maps a GitLab API error onto the sentinels a watcher acts on, leaving everything else untouched so it is treated as transient.
Reader implementations call this so the manager itself stays free of GitLab specifics: it decides *what to do* about an unreadable resource, while this decides *what the failure was*.
The mapping is deliberately coarse. 401, 403 and 404 all become ErrInaccessible because GitLab answers 404 for a resource the caller may not see, precisely so it cannot be distinguished from one that does not exist — and a watcher's response is the same either way: stop, rather than keep polling with a token that has lost access. Everything that is not one of these four statuses stays transient on purpose: a 500, a timeout or a severed connection must cost latency, not the subscription, which is the whole reason polling is the authoritative path.
Types ¶
type Kind ¶
type Kind uint8
Kind classifies a subscribable resource. It determines how a watcher detects change, how fast it polls, and when it retires itself — a pipeline that reached "success" will never change again, whereas a wiki page can change at any time.
const ( // KindUnknown is the zero value, returned for any URI that is not // subscribable. Callers must treat it as "reject", never as a default. KindUnknown Kind = iota // Project-scoped. KindProject // gitlab://project/{ref} KindPipeline // gitlab://project/{ref}/pipeline/{id} KindPipelineJobs // gitlab://project/{ref}/pipeline/{id}/jobs KindPipelineLatest // gitlab://project/{ref}/pipelines/latest KindJob // gitlab://project/{ref}/job/{id} KindMergeRequest // gitlab://project/{ref}/mr/{iid} KindMergeRequestDiscussions // gitlab://project/{ref}/mr/{iid}/discussions KindMergeRequestNotes // gitlab://project/{ref}/mr/{iid}/notes KindIssue // gitlab://project/{ref}/issue/{iid} KindDeployment // gitlab://project/{ref}/deployment/{id} KindEnvironment // gitlab://project/{ref}/environment/{id} KindFeatureFlag // gitlab://project/{ref}/feature_flag/{name} KindRelease // gitlab://project/{ref}/release/{tag} KindTag // gitlab://project/{ref}/tag/{tag} KindBranch // gitlab://project/{ref}/branch/{branch} KindMilestone // gitlab://project/{ref}/milestone/{iid} KindLabel // gitlab://project/{ref}/label/{id} KindBoard // gitlab://project/{ref}/board/{id} KindDeployKey // gitlab://project/{ref}/deploy_key/{id} KindProjectSnippet // gitlab://project/{ref}/snippet/{id} KindWiki // gitlab://project/{ref}/wiki/{slug} KindFile // gitlab://project/{ref}/file/{gitref}/{path} // Group-scoped. KindGroup // gitlab://group/{ref} KindGroupLabel // gitlab://group/{ref}/label/{id} KindGroupMilestone // gitlab://group/{ref}/milestone/{iid} // Instance-scoped. KindSnippet // gitlab://snippet/{id} )
The subscribable resource kinds, one per registered resource template whose content can change over time.
Two categories are deliberately absent. Project- and group-wide collections (issues, branches, labels, members, milestones, releases, tags, projects) are excluded because any change anywhere in the namespace invalidates them, which turns one subscription into a notification firehose and burns the polling budget for no added signal. Commits are excluded because they are immutable: every field a commit resource returns is a property of the commit object itself, which git guarantees is fixed for a given SHA, so a watcher would poll forever and never notify.
func Classify ¶
Classify reports which subscribable resource a concrete URI names, and whether it is subscribable at all.
This is the whitelist the MCP SubscribeHandler enforces. The SDK does not check that a subscribed URI corresponds to any registered resource, and at least one shipping client (Cursor) sends resources/subscribe even against a server that advertises subscribe: false — so rejecting here is what stops clients from holding silent subscriptions to URIs that will never produce a notification.
The whitelist deliberately mirrors what the resource router can actually resolve, rather than being merely a superset of it. Accepting a URI the router would answer with "resource not found" is the worst outcome available: the subscription is acknowledged, then every poll fails, so the client waits for a notification that can never arrive.
func (Kind) String ¶
String returns the kind's name, for logs and error messages. Unmapped values render "unknown" rather than an empty string so a log line never silently loses the field.
func (Kind) Template ¶
Template returns the URI template this kind's resource is registered under, or "" for KindUnknown.
type Manager ¶
type Manager[S comparable] struct { // contains filtered or unexported fields }
Manager owns the watchers for one MCP server, which in HTTP mode means one GitLab token: the server pool keys an *mcp.Server by token and URL, so a manager never spans two identities and a rate-limit pause or an access revocation applies cleanly to everything it owns.
func New ¶
func New[S comparable](reader Reader, notifier Notifier, opts Options) *Manager[S]
New creates a manager. Call Manager.Close to stop every watcher; the manager is unusable afterwards.
func (*Manager[S]) Close ¶
func (m *Manager[S]) Close()
Close stops every watcher and waits for them to finish. It is safe to call more than once.
This is the explicit stop, for an owner shutting the manager down. It is not what bounds a watcher in normal operation: a subscription ends when the session that asked for it disconnects, slows down when its lease runs out unrenewed, and stops for good at Options.MaxLifetime or the first read that says the resource is gone. In HTTP mode the server pool never calls this — it lets an evicted entry expire with its sessions rather than terminating one that may still be serving a live connection, and by the time the last session on that server ends there is nothing left to watch.
func (*Manager[S]) Renew ¶
Renew extends the lease on one URI, restoring full-speed polling if it had slowed. It reports whether anything was watching that URI.
func (*Manager[S]) RenewAll ¶
RenewAll extends the lease on every watch one subscriber holds, and reports how many of them that revived from a demoted state.
This is what ties a subscription's lifetime to its subscriber being present: any request on a session renews everything that session could be waiting on, and nothing else. Renewing per-URI on reads of that URI would be worse than useless — a watcher only notifies on a real change, so a quiet resource produces no notification, no re-read, and would expire during exactly the wait its subscriber cared about. Renewing every watch on the manager would be wrong in the other direction: one busy session would keep another session's abandoned watches at full speed forever.
func (*Manager[S]) Subscribe ¶
Subscribe starts watching uri on behalf of subscriber, or joins the watcher already watching it.
The first subscriber's read happens synchronously, on the caller's context, and doubles as the authorization check: it runs with the subscriber's own token, so a URI this token cannot read is refused here rather than being accepted and then failing silently forever. It also establishes the baseline that later polls are compared against, so a subscriber is never notified about a change that predates its subscription.
A later subscriber waits for that read rather than joining blind, and receives the same answer. It does not read again: every session on one manager shares one token by construction — see Manager — so the first check answered the question for all of them. What it must not do is return success while the only read anyone attempted was still in flight, or had already failed.
Subscribing twice to the same URI as the same subscriber is a no-op: the caller is asking for a state it already holds.
func (*Manager[S]) Unsubscribe ¶
Unsubscribe drops one subscriber's interest in uri. The watcher stops once the last subscriber leaves.
Unsubscribing something this subscriber does not hold is not an error and has no effect: a client may unsubscribe twice, or after a watch already ended on its own, and neither may disturb a watch somebody else is holding.
func (*Manager[S]) UnsubscribeAll ¶
UnsubscribeAll drops every subscription one subscriber holds and reports how many watchers that stopped.
This is what a closing MCP session calls. The SDK drops a disconnected session from its own subscriber table without ever invoking the unsubscribe handler, so without this a watch would outlive the only client that could receive its notifications.
type Notifier ¶
type Notifier interface {
// ResourceUpdated announces that a resource's content changed. The
// error is advisory: the watcher logs it and carries on.
ResourceUpdated(ctx context.Context, update Update) error
}
Notifier delivers a resources/updated notification. Delivery is best-effort, matching MCP's own posture on notifications: a failure is logged and the watcher carries on.
type Options ¶
type Options struct {
// BaseInterval is the polling interval for a resource with no
// lifecycle signal. Defaults to [DefaultBaseInterval].
BaseInterval time.Duration
// MinInterval is the floor for a resource with work in flight.
// Defaults to [DefaultMinInterval].
MinInterval time.Duration
// Lease is how long a subscription is polled at full speed before it
// slows down. Defaults to [DefaultLease].
//
// Reaching it demotes the watcher to [Options.SlowInterval]; it never
// stops one. That distinction is the whole design: MCP has no message
// that means "your subscription expired" — the specification defines no
// lease, no TTL and no renewal, and the one notification available says
// only "this resource changed, read it again" — so a watcher that
// retired at the deadline would go silent in a way no client could tell
// apart from "nothing has happened yet". Slowing down is a claim the
// server can make honestly without saying anything at all.
//
// Any activity on the same session renews it, and so does
// [Manager.Renew].
Lease time.Duration
// SlowInterval is the cadence of a demoted watcher. Defaults to
// [DefaultSlowInterval].
//
// At ten minutes, ten abandoned subscriptions cost one request a
// minute against the 120-a-minute budget of a throttled self-managed
// instance — cheap enough to leave running, slow enough that nobody
// would rely on it, which is what makes a renewal worth asking for.
SlowInterval time.Duration
// MaxLifetime is the absolute cap on a single subscription, renewals
// included. Defaults to [DefaultMaxLifetime]. This is the only deadline
// that truly stops a watcher on time alone.
MaxLifetime time.Duration
// MaxWatchers caps concurrent subscriptions. Defaults to
// [DefaultMaxWatchers]. This is the real safety valve on API budget,
// and it also bounds concurrent outbound requests, which nothing else
// in this server does.
MaxWatchers int
// OnStop, if set, is called once when a watch ends for a reason the
// subscriber did not ask for: [ErrInaccessible], [ErrLifetimeExceeded]
// or [ErrEvicted]. It is not called when a client unsubscribes, when
// its session ends, or when the manager is closed — in all three the
// client either asked for it or is already gone.
//
// It exists so the transport layer can tell the client something, and
// runs on the watcher's goroutine with no locks held.
OnStop func(uri string, reason error)
// Logger defaults to slog.Default().
Logger *slog.Logger
}
Options configures a Manager. The zero value is usable: every field falls back to its documented default.
type Reader ¶
type Reader interface {
// Read returns the current content of uri, or an error describing why
// it could not be read.
Read(ctx context.Context, uri string) ([]byte, error)
}
Reader reads the current content of a resource URI. Implementations return ErrInaccessible or ErrRateLimited where those apply; any other error is treated as transient.
type Update ¶
type Update struct {
// URI is the resource whose content changed.
URI string
// Slow reports that the watch is past its lease and now polling at
// [Options.SlowInterval].
Slow bool
// RenewBy is when the current lease runs out and the watch slows down.
// Any request on the session renews it.
RenewBy time.Time
// Interval is the cadence the watch is running at right now.
Interval time.Duration
}
Update describes a change worth telling a subscriber about, together with the state of the watch that noticed it.
The watch state travels with the notification because there is nowhere else to put it: MCP defines no message for "your subscription slowed down" or "renew it by this time", and a notification is the only thing this server sends unprompted. A client that ignores all of it — every client known today does — still gets a correct notification.
Source Files