Skip to content

Commit 6c7ab45

Browse files
vbilopavclaude
vbilopav
and
claude
committed
feat: SSE @sse_publish/@sse_subscribe split + missed-publish warning
Splits the all-in-one @sse annotation into two independent halves so a routine can be a publisher OR a subscriber without being forced into both. Fully back-compat — existing @sse keeps both halves. @sse_publish - this routine's RAISE feeds the SSE broadcaster; no subscribe URL exposed (cleans up phantom /info URLs on emitter procedures + drops unused createXEventSource helpers from the TS client). @sse_subscribe - exposes a subscribe URL for EventSource clients; routine body is never executed when a client opens the stream. @sse [path] - shorthand for both; unchanged. Adds a runtime warning when a RAISE whose severity matches the configured SSE forwarding level fires in a routine that has no @sse or @sse_publish — kills the silent-emitter footgun where forgetting the annotation produces no error and no event delivery. Once-per-endpoint dedupe; gated on the project actually using SSE somewhere; configurable via the new WarnUnboundServerSentEventsNotices setting (default true). Adds a "connected" handshake flush so SSE clients learn the connection is established before any real event fires — makes "subscribe then publish" sequencing reliable. 12 new tests including a reusable SseTestClient streaming-HTTP helper that opens a connection, waits for the broadcaster to register the subscriber, then reads events. End-to-end live delivery test verifies RAISE in a publish-only procedure reaches a subscriber connected through a separate subscribe-only procedure's URL. 1949/1949 tests pass. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent b2181e2 commit 6c7ab45

23 files changed

Lines changed: 963 additions & 78 deletions

‎NpgsqlRest/Broadcaster.cs‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,4 +47,11 @@ public void CompleteAll()
4747
}
4848
_channels.Clear();
4949
}
50+
51+
/// <summary>
52+
/// Number of currently subscribed channels. Used by integration tests to wait until an SSE
53+
/// subscriber has registered before triggering a publish, avoiding a race between
54+
/// <c>Subscribe</c> and the test's HTTP call. Cheap on a <see cref="ConcurrentDictionary{TKey,TValue}"/>.
55+
/// </summary>
56+
public int SubscriberCount => _channels.Count;
5057
}

‎NpgsqlRest/Defaults/CommentParsers/SseEventsPathHandler.cs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,9 @@ private static void HandleSseEventsPath(
2323
int len,
2424
string description)
2525
{
26+
// @sse is shorthand for "publish from this routine's RAISE AND expose a subscribe URL on the
27+
// same path." See SseEventsPublishHandler / SseEventsSubscribeHandler for the decomposed forms.
28+
endpoint.SsePublishEnabled = true;
2629
if (len == 1)
2730
{
2831
endpoint.SseEventsPath =

‎NpgsqlRest/Defaults/CommentParsers/SseEventsPublishHandler.cs‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
namespace NpgsqlRest.Defaults;
2+
3+
internal static partial class DefaultCommentParser
4+
{
5+
/// <summary>
6+
/// Annotation: sse_publish | sse_events_publish
7+
/// Syntax: sse_publish
8+
/// sse_publish on [info | notice | warning]
9+
///
10+
/// Description: Forward this routine's <c>RAISE</c> statements to the SSE broadcaster — the
11+
/// publish half of <c>@sse</c>. Does NOT expose a subscribe URL on this routine's path; pair
12+
/// with an <c>@sse_subscribe</c> routine elsewhere (or rely on existing subscribers connecting
13+
/// via other subscribe URLs sharing the broadcaster).
14+
/// </summary>
15+
private static readonly string[] SseEventsPublishKey = [
16+
"sse_publish",
17+
"sse_events_publish",
18+
];
19+
20+
private static void HandleSseEventsPublish(
21+
RoutineEndpoint endpoint,
22+
string[] wordsLower,
23+
string[] words,
24+
int len,
25+
string description)
26+
{
27+
endpoint.SsePublishEnabled = true;
28+
29+
if (len >= 3 && StrEquals(wordsLower[1], "on"))
30+
{
31+
if (Enum.TryParse<PostgresNoticeLevels>(words[2], true, out var parsedLevel))
32+
{
33+
endpoint.SseEventNoticeLevel = parsedLevel;
34+
CommentLogger?.CommentSseStreamingLevel(description, endpoint.SseEventNoticeLevel.Value);
35+
}
36+
else
37+
{
38+
Logger?.LogError("Could not recognize valid value for parameter key {key}. Valid values are: {values}. Provided value is {provided}.",
39+
wordsLower[0], string.Join(", ", Enum.GetNames<PostgresNoticeLevels>()), description);
40+
}
41+
}
42+
}
43+
}

‎NpgsqlRest/Defaults/CommentParsers/SseEventsSubscribeHandler.cs‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
namespace NpgsqlRest.Defaults;
2+
3+
internal static partial class DefaultCommentParser
4+
{
5+
/// <summary>
6+
/// Annotation: sse_subscribe | sse_events_subscribe
7+
/// Syntax: sse_subscribe
8+
/// sse_subscribe [path]
9+
/// sse_subscribe [path] on [info | notice | warning]
10+
///
11+
/// Description: Expose an SSE subscribe URL for this routine — the subscribe half of <c>@sse</c>.
12+
/// The routine's body is NEVER executed when a client opens an EventSource against the URL; it
13+
/// only registers the path and (optionally) the notice level filter on outgoing events. Pair
14+
/// with <c>@sse_publish</c> routines (or other <c>@sse</c> routines) elsewhere to source the
15+
/// events that flow through this URL.
16+
/// </summary>
17+
private static readonly string[] SseEventsSubscribeKey = [
18+
"sse_subscribe",
19+
"sse_events_subscribe",
20+
];
21+
22+
private static void HandleSseEventsSubscribe(
23+
RoutineEndpoint endpoint,
24+
string[] wordsLower,
25+
string[] words,
26+
int len,
27+
string description)
28+
{
29+
// Subscribe-only: URL exposed but RAISE in this routine's body is NOT forwarded. Don't touch
30+
// SsePublishEnabled — leaving it false is the whole point of the decomposition.
31+
if (len == 1)
32+
{
33+
endpoint.SseEventsPath =
34+
(endpoint.SseEventNoticeLevel ?? Options.DefaultSseEventNoticeLevel).ToString().ToLowerInvariant();
35+
CommentLogger?.CommentSseStreamingPath(description, endpoint.SseEventsPath);
36+
}
37+
else
38+
{
39+
endpoint.SseEventsPath = wordsLower[1];
40+
if (len >= 4 && StrEquals(wordsLower[2], "on"))
41+
{
42+
if (Enum.TryParse<PostgresNoticeLevels>(words[3], true, out var parsedLevel))
43+
{
44+
endpoint.SseEventNoticeLevel = parsedLevel;
45+
CommentLogger?.CommentSseStreamingPathAndLevel(description, endpoint.SseEventsPath, endpoint.SseEventNoticeLevel.Value);
46+
}
47+
else
48+
{
49+
Logger?.LogError("Could not recognize valid value for parameter key {key}. Valid values are: {values}. Provided value is {provided}.",
50+
wordsLower[0], string.Join(", ", Enum.GetNames<PostgresNoticeLevels>()), description);
51+
}
52+
}
53+
else
54+
{
55+
CommentLogger?.CommentSseStreamingPath(description, endpoint.SseEventsPath);
56+
}
57+
}
58+
}
59+
}

‎NpgsqlRest/Defaults/DefaultCommentParser.cs‎

Lines changed: 47 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -366,6 +366,22 @@ internal static partial class DefaultCommentParser
366366
TrackAnnotation(line);
367367
}
368368

369+
// sse_publish [ on info | notice | warning ]
370+
// sse_events_publish [ on info | notice | warning ]
371+
else if (haveTag is true && StrEqualsToArray(wordsLower[0], SseEventsPublishKey))
372+
{
373+
HandleSseEventsPublish(routineEndpoint, wordsLower, words, len, description);
374+
TrackAnnotation(line);
375+
}
376+
377+
// sse_subscribe [ path ] [ on info | notice | warning ]
378+
// sse_events_subscribe [ path ] [ on info | notice | warning ]
379+
else if (haveTag is true && StrEqualsToArray(wordsLower[0], SseEventsSubscribeKey))
380+
{
381+
HandleSseEventsSubscribe(routineEndpoint, wordsLower, words, len, description);
382+
TrackAnnotation(line);
383+
}
384+
369385
// sse_level [ info | notice | warning ]
370386
// sse_events_level [ info | notice | warning ]
371387
else if (haveTag is true && len >= 2 && StrEqualsToArray(wordsLower[0], SseEventsLevelKey))
@@ -598,10 +614,39 @@ public static void SetCustomParameter(RoutineEndpoint endpoint, string name, str
598614

599615
else if (StrEqualsToArray(name, SseEventsStreamingPathKey))
600616
{
617+
// Tag-value form of @sse: same shorthand semantics — publish + subscribe on the same path.
601618
if (bool.TryParse(value, out var parseredStreamingPath))
602619
{
603-
endpoint.SseEventsPath = parseredStreamingPath is true ?
604-
(endpoint.SseEventNoticeLevel ?? Options.DefaultSseEventNoticeLevel).ToString().ToLowerInvariant() :
620+
endpoint.SseEventsPath = parseredStreamingPath is true ?
621+
(endpoint.SseEventNoticeLevel ?? Options.DefaultSseEventNoticeLevel).ToString().ToLowerInvariant() :
622+
null;
623+
endpoint.SsePublishEnabled = parseredStreamingPath;
624+
}
625+
else
626+
{
627+
endpoint.SseEventsPath = value;
628+
endpoint.SsePublishEnabled = true;
629+
}
630+
}
631+
632+
else if (StrEqualsToArray(name, SseEventsPublishKey))
633+
{
634+
if (bool.TryParse(value, out var parsedPublish))
635+
{
636+
endpoint.SsePublishEnabled = parsedPublish;
637+
}
638+
else
639+
{
640+
endpoint.SsePublishEnabled = true;
641+
}
642+
}
643+
644+
else if (StrEqualsToArray(name, SseEventsSubscribeKey))
645+
{
646+
if (bool.TryParse(value, out var parsedSubscribe))
647+
{
648+
endpoint.SseEventsPath = parsedSubscribe is true ?
649+
(endpoint.SseEventNoticeLevel ?? Options.DefaultSseEventNoticeLevel).ToString().ToLowerInvariant() :
605650
null;
606651
}
607652
else

‎NpgsqlRest/Log.cs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -334,4 +334,7 @@ public static partial class Log
334334

335335
[LoggerMessage(Level = LogLevel.Warning, Message = "Endpoint {path} parameter {paramName} received a {source} value but is auto-bound from claim '{claimName}'. The supplied value is being ignored.")]
336336
public static partial void ClaimMappedParamReceivedRequestValue(this ILogger logger, string path, string paramName, string source, string claimName);
337+
338+
[LoggerMessage(Level = LogLevel.Warning, Message = "RAISE {severity} in endpoint {path} was not broadcast to SSE subscribers — the endpoint has no @sse or @sse_publish annotation. Add @sse_publish to forward this routine's notices, or set NpgsqlRestOptions.WarnUnboundSseNotices=false to silence this warning.")]
339+
public static partial void UnboundSseRaise(this ILogger logger, string severity, string path);
337340
}

‎NpgsqlRest/NpgsqlRestBuilder.cs‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -133,13 +133,21 @@ public static IApplicationBuilder UseNpgsqlRest(this WebApplication builder, Npg
133133
{
134134
Logger?.EndpointCreated(urlInfo);
135135
}
136-
if (endpoint.SseEventsPath is not null)
136+
// Resolve the notice-level default for any endpoint that participates in SSE — both
137+
// subscribers (need it for filtering) and publishers (need it for the broadcast gate).
138+
// After B.1 these can be independent: an @sse_publish-only endpoint has no SseEventsPath
139+
// but still needs a configured level; an @sse_subscribe-only endpoint has a path but
140+
// doesn't publish.
141+
if (endpoint.SseEventsPath is not null || endpoint.SsePublishEnabled)
137142
{
138143
if (endpoint.SseEventNoticeLevel is null)
139144
{
140145
endpoint.SseEventNoticeLevel = Options.DefaultSseEventNoticeLevel;
141146
}
142-
Logger?.EndpointSsePath($"{endpoint.Method} {endpoint.Path}", endpoint.SseEventsPath, endpoint.SseEventNoticeLevel);
147+
if (endpoint.SseEventsPath is not null)
148+
{
149+
Logger?.EndpointSsePath($"{endpoint.Method} {endpoint.Path}", endpoint.SseEventsPath, endpoint.SseEventNoticeLevel);
150+
}
143151
}
144152
}
145153

@@ -384,6 +392,16 @@ private static Metadata Build(IApplicationBuilder? builder, RetryStrategy? defau
384392
}
385393

386394
NpgsqlRestSseEventSource.Paths.Add(endpoint.SseEventsPath);
395+
NpgsqlRestSseEventSource.HasAnySseEndpoints = true;
396+
hasStreamingEvents = true;
397+
}
398+
399+
// Track whether anything in this build publishes to the broadcaster — both feeds the
400+
// SSE middleware and (combined with subscribe registrations above) gates the
401+
// unbound-RAISE warning so projects that don't use SSE see zero noise.
402+
if (endpoint.SsePublishEnabled)
403+
{
404+
NpgsqlRestSseEventSource.HasAnySseEndpoints = true;
387405
hasStreamingEvents = true;
388406
}
389407

‎NpgsqlRest/NpgsqlRestEndpoint.cs‎

Lines changed: 25 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -122,22 +122,41 @@ public async Task InvokeAsync(HttpContext context, IServiceProvider serviceProvi
122122
return;
123123
}
124124

125-
if ( (Options.LogConnectionNoticeEvents && Logger != null) || endpoint.SseEventsPath is not null)
125+
// Three reasons to attach a notice handler:
126+
// 1. LogConnectionNoticeEvents — log every notice regardless of SSE wiring.
127+
// 2. Endpoint publishes to broadcaster — forward matching notices to subscribers.
128+
// 3. WarnUnboundSseNotices — and the project actually uses SSE somewhere; warn when a
129+
// RAISE in a non-publishing endpoint matches the SSE forwarding level (a likely
130+
// missing @sse_publish annotation).
131+
bool noticeWarnEnabled = Options.WarnUnboundSseNotices
132+
&& Logger != null
133+
&& NpgsqlRestSseEventSource.HasAnySseEndpoints
134+
&& endpoint.SsePublishEnabled is false;
135+
136+
if ((Options.LogConnectionNoticeEvents && Logger != null)
137+
|| endpoint.SsePublishEnabled
138+
|| noticeWarnEnabled)
126139
{
127-
var currentEndpoint = endpoint;
140+
var currentEndpointCaptured = endpoint;
128141
connection.Notice += (sender, args) =>
129142
{
130143
if (Options.LogConnectionNoticeEvents && Logger != null)
131144
{
132145
NpgsqlRestLogger.LogConnectionNotice(args.Notice, Options.LogConnectionNoticeEventsMode);
133146
}
134-
if (currentEndpoint.SseEventsPath is not null &&
135-
currentEndpoint.SseEventNoticeLevel is not null &&
136-
string.Equals(args.Notice.Severity, currentEndpoint.SseEventNoticeLevel.ToString(), StringComparison.OrdinalIgnoreCase))
147+
if (currentEndpointCaptured.SsePublishEnabled
148+
&& currentEndpointCaptured.SseEventNoticeLevel is not null
149+
&& string.Equals(args.Notice.Severity, currentEndpointCaptured.SseEventNoticeLevel.ToString(), StringComparison.OrdinalIgnoreCase))
137150
{
138151
NpgsqlRestSseEventSource
139152
.Broadcaster
140-
.Broadcast(new SseEvent(args.Notice, currentEndpoint, context.Request.Headers[Options.ExecutionIdHeaderName].FirstOrDefault()));
153+
.Broadcast(new SseEvent(args.Notice, currentEndpointCaptured, context.Request.Headers[Options.ExecutionIdHeaderName].FirstOrDefault()));
154+
}
155+
else if (noticeWarnEnabled
156+
&& string.Equals(args.Notice.Severity, Options.DefaultSseEventNoticeLevel.ToString(), StringComparison.OrdinalIgnoreCase)
157+
&& SseUnboundWarner.TryMarkWarned(currentEndpointCaptured.Path))
158+
{
159+
Logger!.UnboundSseRaise(args.Notice.Severity, currentEndpointCaptured.Path);
141160
}
142161
};
143162
}

‎NpgsqlRest/NpgsqlRestSseEventSource.cs‎

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,15 @@ public class NpgsqlRestSseEventSource(RequestDelegate next)
1212
{
1313
public static readonly HashSet<string> Paths = [];
1414
public static readonly Broadcaster<SseEvent> Broadcaster = new();
15+
16+
/// <summary>
17+
/// True if any endpoint in the build participates in SSE — either as a publisher (<c>@sse</c> /
18+
/// <c>@sse_publish</c>) or as a subscribe URL (<c>@sse</c> / <c>@sse_subscribe</c>). Used as a fast
19+
/// guard for the unbound-RAISE warning so projects that don't use SSE at all pay zero per-notice
20+
/// cost. Once any endpoint registers SSE usage, every other endpoint's notices are eligible to
21+
/// trigger the warning when their <c>RAISE</c> level matches the configured forwarding level.
22+
/// </summary>
23+
public static bool HasAnySseEndpoints { get; internal set; } = false;
1524

1625
public async Task InvokeAsync(HttpContext context)
1726
{
@@ -45,6 +54,23 @@ public async Task InvokeAsync(HttpContext context)
4554

4655
var connectionId = Guid.NewGuid();
4756
var reader = Broadcaster.Subscribe(connectionId);
57+
// Force the response headers + an initial body byte to flush immediately. SSE permits
58+
// comment-only lines (lines starting with ':'); browsers and EventSource clients ignore
59+
// them. Without an early flush, the GET stays "headers pending" until the first real event
60+
// — which prevents callers (notably integration tests) from sequencing "subscribe THEN
61+
// publish": their GET wouldn't return until an event had already been received, and the
62+
// server's await-foreach below blocks waiting for that same event. Writing a small comment
63+
// breaks the deadlock and acts as a connection-established ping.
64+
try
65+
{
66+
await context.Response.WriteAsync(": connected\n\n", cancellationToken);
67+
await context.Response.Body.FlushAsync(cancellationToken);
68+
}
69+
catch (OperationCanceledException)
70+
{
71+
Broadcaster.Unsubscribe(connectionId);
72+
return;
73+
}
4874
try
4975
{
5076
await foreach (var noticeEvent in reader.ReadAllAsync(cancellationToken))

‎NpgsqlRest/Options/NpgsqlRestOptions.cs‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -368,6 +368,15 @@ public NpgsqlRestOptions(NpgsqlDataSource dataSource)
368368
/// Collection of custom server-sent events response headers that will be added to the response when connected to the endpoint that is configured to return server-sent events.
369369
/// </summary>
370370
public Dictionary<string, StringValues> SseResponseHeaders { get; set; } = [];
371+
372+
/// <summary>
373+
/// When true (default), log a one-time warning per endpoint when a <c>RAISE</c> at the configured
374+
/// SSE notice level fires inside a routine that has no <c>@sse</c> or <c>@sse_publish</c>
375+
/// annotation. The notice is logged but is NOT broadcast to SSE subscribers; the warning surfaces
376+
/// the likely missing annotation. Inactive when no endpoint in the build participates in SSE
377+
/// publishing — so projects that don't use SSE pay zero overhead and see no warnings.
378+
/// </summary>
379+
public bool WarnUnboundSseNotices { get; set; } = true;
371380

372381
/// <summary>
373382
/// Default rate limiting policy for all requests. Policy must be configured within application rate limiting options.

0 commit comments

Comments
 (0)