From 4738b3bc7f027bca7d64204fb35eaf87e48f3ca5 Mon Sep 17 00:00:00 2001
From: Moe Dashti <mohammad.dashti@temporal.io>
Date: Tue, 13 Oct 2026 18:57:22 -0700
Subject: [PATCH] Started a missing notifier on a notification and added the
 poll.

A reader that armed the store may reach the notifier after the producer's notification, so the notification keeps the counter it waits for. A client waits on PollStreamNotifier in place of a blocking store read.
---
 chasm/lib/streamnotifier/config.go            |  62 ++++---
 chasm/lib/streamnotifier/frontend.go          |  23 +++
 .../v1/request_response.go-helpers.pb.go      |  74 ++++++++
 .../v1/request_response.pb.go                 | 165 +++++++++++++++---
 .../gen/streamnotifierpb/v1/service.pb.go     |  29 +--
 .../streamnotifierpb/v1/service_client.pb.go  |  43 +++++
 .../streamnotifierpb/v1/service_grpc.pb.go    |  37 ++++
 chasm/lib/streamnotifier/handler.go           | 128 +++++++++++---
 chasm/lib/streamnotifier/notifier.go          |  40 ++++-
 chasm/lib/streamnotifier/notifier_test.go     | 163 ++++++++++++++---
 .../proto/v1/request_response.proto           |  10 ++
 .../lib/streamnotifier/proto/v1/service.proto |   5 +
 .../rpc/interceptor/dc_redirection_policy.go  |   1 +
 .../interceptor/dc_redirection_policy_test.go |   1 +
 service/frontend/stream_notifier_handler.go   |  16 --
 tests/stream_poll_test.go                     | 136 +++++++++++++++
 16 files changed, 814 insertions(+), 119 deletions(-)
 delete mode 100644 service/frontend/stream_notifier_handler.go
 create mode 100644 tests/stream_poll_test.go

diff --git a/chasm/lib/streamnotifier/config.go b/chasm/lib/streamnotifier/config.go
index 93c80ab4001..7f1f41663dc 100644
--- a/chasm/lib/streamnotifier/config.go
+++ b/chasm/lib/streamnotifier/config.go
@@ -46,6 +46,24 @@ open: later notifications and attaches work as before. A notifier that goes idle
 ends, and the next attach or notification starts a new one.`,
 )
 
+var IdleTimeoutNoCallbacks = dynamicconfig.NewNamespaceDurationSetting(
+	"streamnotifier.idleTimeoutNoCallbacks",
+	time.Hour,
+	`How long an open stream notifier with no callbacks waiting waits for an attach or a
+notification before it ends. A notification or a poll starts a notifier when the stream has none,
+so a sleeping reader has a notifier to wake it. A notification that names the wrong first run
+starts one that nobody attaches to, and this ends it soon.`,
+)
+
+var PollMaxWait = dynamicconfig.NewNamespaceDurationSetting(
+	"streamnotifier.pollMaxWait",
+	time.Minute,
+	`The longest PollStreamNotifier waits for a counter above the one it names, or for the close.
+At the limit it returns with no change, and the client reads the stream again. That read finds
+records whose notification was lost, so this also bounds how long a client sleeps after a
+producer stopped before it notified.`,
+)
+
 var OwnerCheckInterval = dynamicconfig.NewNamespaceDurationSetting(
 	"streamnotifier.ownerCheckInterval",
 	5*time.Minute,
@@ -75,15 +93,17 @@ still holds the records it would read. The default matches the default store ret
 
 // Config holds the stream notifier settings.
 type Config struct {
-	Enabled            dynamicconfig.BoolPropertyFnWithNamespaceFilter
-	ChasmEnabled       dynamicconfig.BoolPropertyFnWithNamespaceFilter
-	MaxCallbacks       dynamicconfig.IntPropertyFnWithNamespaceFilter
-	MaxCallbackBytes   dynamicconfig.IntPropertyFnWithNamespaceFilter
-	IdleTimeout        dynamicconfig.DurationPropertyFnWithNamespaceFilter
-	ClosedRetention    dynamicconfig.DurationPropertyFnWithNamespaceFilter
-	OwnerCheckInterval dynamicconfig.DurationPropertyFnWithNamespaceFilter
-	OwnerCloseGrace    dynamicconfig.DurationPropertyFnWithNamespaceFilter
-	MaxIDLength        dynamicconfig.IntPropertyFn
+	Enabled                dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	ChasmEnabled           dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	MaxCallbacks           dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxCallbackBytes       dynamicconfig.IntPropertyFnWithNamespaceFilter
+	IdleTimeout            dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	IdleTimeoutNoCallbacks dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	PollMaxWait            dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	ClosedRetention        dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	OwnerCheckInterval     dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	OwnerCloseGrace        dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	MaxIDLength            dynamicconfig.IntPropertyFn
 	// The close result becomes the operation's result, so it takes the payload blob limits.
 	BlobSizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter
 	BlobSizeLimitWarn  dynamicconfig.IntPropertyFnWithNamespaceFilter
@@ -91,16 +111,18 @@ type Config struct {
 
 func configProvider(dc *dynamicconfig.Collection) *Config {
 	return &Config{
-		Enabled:            Enabled.Get(dc),
-		ChasmEnabled:       dynamicconfig.EnableChasm.Get(dc),
-		MaxCallbacks:       MaxCallbacks.Get(dc),
-		MaxCallbackBytes:   MaxCallbackBytes.Get(dc),
-		IdleTimeout:        IdleTimeout.Get(dc),
-		ClosedRetention:    ClosedRetention.Get(dc),
-		OwnerCheckInterval: OwnerCheckInterval.Get(dc),
-		OwnerCloseGrace:    OwnerCloseGrace.Get(dc),
-		MaxIDLength:        dynamicconfig.MaxIDLengthLimit.Get(dc),
-		BlobSizeLimitError: dynamicconfig.BlobSizeLimitError.Get(dc),
-		BlobSizeLimitWarn:  dynamicconfig.BlobSizeLimitWarn.Get(dc),
+		Enabled:                Enabled.Get(dc),
+		ChasmEnabled:           dynamicconfig.EnableChasm.Get(dc),
+		MaxCallbacks:           MaxCallbacks.Get(dc),
+		MaxCallbackBytes:       MaxCallbackBytes.Get(dc),
+		IdleTimeout:            IdleTimeout.Get(dc),
+		IdleTimeoutNoCallbacks: IdleTimeoutNoCallbacks.Get(dc),
+		PollMaxWait:            PollMaxWait.Get(dc),
+		ClosedRetention:        ClosedRetention.Get(dc),
+		OwnerCheckInterval:     OwnerCheckInterval.Get(dc),
+		OwnerCloseGrace:        OwnerCloseGrace.Get(dc),
+		MaxIDLength:            dynamicconfig.MaxIDLengthLimit.Get(dc),
+		BlobSizeLimitError:     dynamicconfig.BlobSizeLimitError.Get(dc),
+		BlobSizeLimitWarn:      dynamicconfig.BlobSizeLimitWarn.Get(dc),
 	}
 }
diff --git a/chasm/lib/streamnotifier/frontend.go b/chasm/lib/streamnotifier/frontend.go
index a1a0f45faab..8eb7be0a578 100644
--- a/chasm/lib/streamnotifier/frontend.go
+++ b/chasm/lib/streamnotifier/frontend.go
@@ -42,6 +42,10 @@ type FrontendHandler interface {
 		context.Context,
 		*workflowservice.DescribeStreamNotifierRequest,
 	) (*workflowservice.DescribeStreamNotifierResponse, error)
+	PollStreamNotifier(
+		context.Context,
+		*workflowservice.PollStreamNotifierRequest,
+	) (*workflowservice.PollStreamNotifierResponse, error)
 }
 
 type frontendHandler struct {
@@ -255,3 +259,22 @@ func (h *frontendHandler) DescribeStreamNotifier(
 	)
 	return resp.GetFrontendResponse(), err
 }
+
+func (h *frontendHandler) PollStreamNotifier(
+	ctx context.Context,
+	req *workflowservice.PollStreamNotifierRequest,
+) (*workflowservice.PollStreamNotifierResponse, error) {
+	namespaceID, businessID, err := h.resolve(req.GetNamespace(), req.GetStreamRef())
+	if err != nil {
+		return nil, err
+	}
+	if req.GetAfterCounter() < 0 {
+		return nil, serviceerror.NewInvalidArgument("after_counter must not be negative")
+	}
+	resp, err := h.client.PollStreamNotifier(ctx, &streamnotifierpb.PollStreamNotifierRequest{
+		NamespaceId:     namespaceID,
+		BusinessId:      businessID,
+		FrontendRequest: req,
+	})
+	return resp.GetFrontendResponse(), err
+}
diff --git a/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/request_response.go-helpers.pb.go b/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/request_response.go-helpers.pb.go
index ab9c0539d66..eaf5752e46e 100644
--- a/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/request_response.go-helpers.pb.go
+++ b/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/request_response.go-helpers.pb.go
@@ -300,3 +300,77 @@ func (this *DescribeStreamNotifierResponse) Equal(that interface{}) bool {
 
 	return proto.Equal(this, that1)
 }
+
+// Marshal an object of type PollStreamNotifierRequest to the protobuf v3 wire format
+func (val *PollStreamNotifierRequest) Marshal() ([]byte, error) {
+	return proto.Marshal(val)
+}
+
+// Unmarshal an object of type PollStreamNotifierRequest from the protobuf v3 wire format
+func (val *PollStreamNotifierRequest) Unmarshal(buf []byte) error {
+	return proto.Unmarshal(buf, val)
+}
+
+// Size returns the size of the object, in bytes, once serialized
+func (val *PollStreamNotifierRequest) Size() int {
+	return proto.Size(val)
+}
+
+// Equal returns whether two PollStreamNotifierRequest values are equivalent by recursively
+// comparing the message's fields.
+// For more information see the documentation for
+// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
+func (this *PollStreamNotifierRequest) Equal(that interface{}) bool {
+	if that == nil {
+		return this == nil
+	}
+
+	var that1 *PollStreamNotifierRequest
+	switch t := that.(type) {
+	case *PollStreamNotifierRequest:
+		that1 = t
+	case PollStreamNotifierRequest:
+		that1 = &t
+	default:
+		return false
+	}
+
+	return proto.Equal(this, that1)
+}
+
+// Marshal an object of type PollStreamNotifierResponse to the protobuf v3 wire format
+func (val *PollStreamNotifierResponse) Marshal() ([]byte, error) {
+	return proto.Marshal(val)
+}
+
+// Unmarshal an object of type PollStreamNotifierResponse from the protobuf v3 wire format
+func (val *PollStreamNotifierResponse) Unmarshal(buf []byte) error {
+	return proto.Unmarshal(buf, val)
+}
+
+// Size returns the size of the object, in bytes, once serialized
+func (val *PollStreamNotifierResponse) Size() int {
+	return proto.Size(val)
+}
+
+// Equal returns whether two PollStreamNotifierResponse values are equivalent by recursively
+// comparing the message's fields.
+// For more information see the documentation for
+// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
+func (this *PollStreamNotifierResponse) Equal(that interface{}) bool {
+	if that == nil {
+		return this == nil
+	}
+
+	var that1 *PollStreamNotifierResponse
+	switch t := that.(type) {
+	case *PollStreamNotifierResponse:
+		that1 = t
+	case PollStreamNotifierResponse:
+		that1 = &t
+	default:
+		return false
+	}
+
+	return proto.Equal(this, that1)
+}
diff --git a/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/request_response.pb.go b/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/request_response.pb.go
index d200f072c43..e56d4baf8bf 100644
--- a/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/request_response.pb.go
+++ b/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/request_response.pb.go
@@ -439,6 +439,110 @@ func (x *DescribeStreamNotifierResponse) GetFrontendResponse() *v1.DescribeStrea
 	return nil
 }
 
+type PollStreamNotifierRequest struct {
+	state           protoimpl.MessageState        `protogen:"open.v1"`
+	NamespaceId     string                        `protobuf:"bytes,1,opt,name=namespace_id,json=namespaceId,proto3" json:"namespace_id,omitempty"`
+	BusinessId      string                        `protobuf:"bytes,2,opt,name=business_id,json=businessId,proto3" json:"business_id,omitempty"`
+	FrontendRequest *v1.PollStreamNotifierRequest `protobuf:"bytes,3,opt,name=frontend_request,json=frontendRequest,proto3" json:"frontend_request,omitempty"`
+	unknownFields   protoimpl.UnknownFields
+	sizeCache       protoimpl.SizeCache
+}
+
+func (x *PollStreamNotifierRequest) Reset() {
+	*x = PollStreamNotifierRequest{}
+	mi := &file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_msgTypes[8]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *PollStreamNotifierRequest) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*PollStreamNotifierRequest) ProtoMessage() {}
+
+func (x *PollStreamNotifierRequest) ProtoReflect() protoreflect.Message {
+	mi := &file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_msgTypes[8]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use PollStreamNotifierRequest.ProtoReflect.Descriptor instead.
+func (*PollStreamNotifierRequest) Descriptor() ([]byte, []int) {
+	return file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_rawDescGZIP(), []int{8}
+}
+
+func (x *PollStreamNotifierRequest) GetNamespaceId() string {
+	if x != nil {
+		return x.NamespaceId
+	}
+	return ""
+}
+
+func (x *PollStreamNotifierRequest) GetBusinessId() string {
+	if x != nil {
+		return x.BusinessId
+	}
+	return ""
+}
+
+func (x *PollStreamNotifierRequest) GetFrontendRequest() *v1.PollStreamNotifierRequest {
+	if x != nil {
+		return x.FrontendRequest
+	}
+	return nil
+}
+
+type PollStreamNotifierResponse struct {
+	state            protoimpl.MessageState         `protogen:"open.v1"`
+	FrontendResponse *v1.PollStreamNotifierResponse `protobuf:"bytes,1,opt,name=frontend_response,json=frontendResponse,proto3" json:"frontend_response,omitempty"`
+	unknownFields    protoimpl.UnknownFields
+	sizeCache        protoimpl.SizeCache
+}
+
+func (x *PollStreamNotifierResponse) Reset() {
+	*x = PollStreamNotifierResponse{}
+	mi := &file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_msgTypes[9]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *PollStreamNotifierResponse) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*PollStreamNotifierResponse) ProtoMessage() {}
+
+func (x *PollStreamNotifierResponse) ProtoReflect() protoreflect.Message {
+	mi := &file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_msgTypes[9]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use PollStreamNotifierResponse.ProtoReflect.Descriptor instead.
+func (*PollStreamNotifierResponse) Descriptor() ([]byte, []int) {
+	return file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_rawDescGZIP(), []int{9}
+}
+
+func (x *PollStreamNotifierResponse) GetFrontendResponse() *v1.PollStreamNotifierResponse {
+	if x != nil {
+		return x.FrontendResponse
+	}
+	return nil
+}
+
 var File_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto protoreflect.FileDescriptor
 
 const file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_rawDesc = "" +
@@ -471,7 +575,14 @@ const file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_pr
 	"businessId\x12i\n" +
 	"\x10frontend_request\x18\x03 \x01(\v2>.temporal.api.workflowservice.v1.DescribeStreamNotifierRequestR\x0ffrontendRequest\"\x8e\x01\n" +
 	"\x1eDescribeStreamNotifierResponse\x12l\n" +
-	"\x11frontend_response\x18\x01 \x01(\v2?.temporal.api.workflowservice.v1.DescribeStreamNotifierResponseR\x10frontendResponseBVZTgo.temporal.io/server/chasm/lib/streamnotifier/gen/streamnotifierpb;streamnotifierpbb\x06proto3"
+	"\x11frontend_response\x18\x01 \x01(\v2?.temporal.api.workflowservice.v1.DescribeStreamNotifierResponseR\x10frontendResponse\"\xc6\x01\n" +
+	"\x19PollStreamNotifierRequest\x12!\n" +
+	"\fnamespace_id\x18\x01 \x01(\tR\vnamespaceId\x12\x1f\n" +
+	"\vbusiness_id\x18\x02 \x01(\tR\n" +
+	"businessId\x12e\n" +
+	"\x10frontend_request\x18\x03 \x01(\v2:.temporal.api.workflowservice.v1.PollStreamNotifierRequestR\x0ffrontendRequest\"\x86\x01\n" +
+	"\x1aPollStreamNotifierResponse\x12h\n" +
+	"\x11frontend_response\x18\x01 \x01(\v2;.temporal.api.workflowservice.v1.PollStreamNotifierResponseR\x10frontendResponseBVZTgo.temporal.io/server/chasm/lib/streamnotifier/gen/streamnotifierpb;streamnotifierpbb\x06proto3"
 
 var (
 	file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_rawDescOnce sync.Once
@@ -485,7 +596,7 @@ func file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_pro
 	return file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_rawDescData
 }
 
-var file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_msgTypes = make([]protoimpl.MessageInfo, 8)
+var file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_msgTypes = make([]protoimpl.MessageInfo, 10)
 var file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_goTypes = []any{
 	(*AttachStreamCallbackRequest)(nil),       // 0: temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackRequest
 	(*AttachStreamCallbackResponse)(nil),      // 1: temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackResponse
@@ -495,29 +606,35 @@ var file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_prot
 	(*NotifyStreamResponse)(nil),              // 5: temporal.server.chasm.lib.streamnotifier.proto.v1.NotifyStreamResponse
 	(*DescribeStreamNotifierRequest)(nil),     // 6: temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierRequest
 	(*DescribeStreamNotifierResponse)(nil),    // 7: temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierResponse
-	(*v1.AttachStreamCallbackRequest)(nil),    // 8: temporal.api.workflowservice.v1.AttachStreamCallbackRequest
-	(*v1.AttachStreamCallbackResponse)(nil),   // 9: temporal.api.workflowservice.v1.AttachStreamCallbackResponse
-	(*v1.DetachStreamCallbackRequest)(nil),    // 10: temporal.api.workflowservice.v1.DetachStreamCallbackRequest
-	(*v1.DetachStreamCallbackResponse)(nil),   // 11: temporal.api.workflowservice.v1.DetachStreamCallbackResponse
-	(*v1.NotifyStreamRequest)(nil),            // 12: temporal.api.workflowservice.v1.NotifyStreamRequest
-	(*v1.NotifyStreamResponse)(nil),           // 13: temporal.api.workflowservice.v1.NotifyStreamResponse
-	(*v1.DescribeStreamNotifierRequest)(nil),  // 14: temporal.api.workflowservice.v1.DescribeStreamNotifierRequest
-	(*v1.DescribeStreamNotifierResponse)(nil), // 15: temporal.api.workflowservice.v1.DescribeStreamNotifierResponse
+	(*PollStreamNotifierRequest)(nil),         // 8: temporal.server.chasm.lib.streamnotifier.proto.v1.PollStreamNotifierRequest
+	(*PollStreamNotifierResponse)(nil),        // 9: temporal.server.chasm.lib.streamnotifier.proto.v1.PollStreamNotifierResponse
+	(*v1.AttachStreamCallbackRequest)(nil),    // 10: temporal.api.workflowservice.v1.AttachStreamCallbackRequest
+	(*v1.AttachStreamCallbackResponse)(nil),   // 11: temporal.api.workflowservice.v1.AttachStreamCallbackResponse
+	(*v1.DetachStreamCallbackRequest)(nil),    // 12: temporal.api.workflowservice.v1.DetachStreamCallbackRequest
+	(*v1.DetachStreamCallbackResponse)(nil),   // 13: temporal.api.workflowservice.v1.DetachStreamCallbackResponse
+	(*v1.NotifyStreamRequest)(nil),            // 14: temporal.api.workflowservice.v1.NotifyStreamRequest
+	(*v1.NotifyStreamResponse)(nil),           // 15: temporal.api.workflowservice.v1.NotifyStreamResponse
+	(*v1.DescribeStreamNotifierRequest)(nil),  // 16: temporal.api.workflowservice.v1.DescribeStreamNotifierRequest
+	(*v1.DescribeStreamNotifierResponse)(nil), // 17: temporal.api.workflowservice.v1.DescribeStreamNotifierResponse
+	(*v1.PollStreamNotifierRequest)(nil),      // 18: temporal.api.workflowservice.v1.PollStreamNotifierRequest
+	(*v1.PollStreamNotifierResponse)(nil),     // 19: temporal.api.workflowservice.v1.PollStreamNotifierResponse
 }
 var file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_depIdxs = []int32{
-	8,  // 0: temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.AttachStreamCallbackRequest
-	9,  // 1: temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackResponse.frontend_response:type_name -> temporal.api.workflowservice.v1.AttachStreamCallbackResponse
-	10, // 2: temporal.server.chasm.lib.streamnotifier.proto.v1.DetachStreamCallbackRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.DetachStreamCallbackRequest
-	11, // 3: temporal.server.chasm.lib.streamnotifier.proto.v1.DetachStreamCallbackResponse.frontend_response:type_name -> temporal.api.workflowservice.v1.DetachStreamCallbackResponse
-	12, // 4: temporal.server.chasm.lib.streamnotifier.proto.v1.NotifyStreamRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.NotifyStreamRequest
-	13, // 5: temporal.server.chasm.lib.streamnotifier.proto.v1.NotifyStreamResponse.frontend_response:type_name -> temporal.api.workflowservice.v1.NotifyStreamResponse
-	14, // 6: temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.DescribeStreamNotifierRequest
-	15, // 7: temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierResponse.frontend_response:type_name -> temporal.api.workflowservice.v1.DescribeStreamNotifierResponse
-	8,  // [8:8] is the sub-list for method output_type
-	8,  // [8:8] is the sub-list for method input_type
-	8,  // [8:8] is the sub-list for extension type_name
-	8,  // [8:8] is the sub-list for extension extendee
-	0,  // [0:8] is the sub-list for field type_name
+	10, // 0: temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.AttachStreamCallbackRequest
+	11, // 1: temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackResponse.frontend_response:type_name -> temporal.api.workflowservice.v1.AttachStreamCallbackResponse
+	12, // 2: temporal.server.chasm.lib.streamnotifier.proto.v1.DetachStreamCallbackRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.DetachStreamCallbackRequest
+	13, // 3: temporal.server.chasm.lib.streamnotifier.proto.v1.DetachStreamCallbackResponse.frontend_response:type_name -> temporal.api.workflowservice.v1.DetachStreamCallbackResponse
+	14, // 4: temporal.server.chasm.lib.streamnotifier.proto.v1.NotifyStreamRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.NotifyStreamRequest
+	15, // 5: temporal.server.chasm.lib.streamnotifier.proto.v1.NotifyStreamResponse.frontend_response:type_name -> temporal.api.workflowservice.v1.NotifyStreamResponse
+	16, // 6: temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.DescribeStreamNotifierRequest
+	17, // 7: temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierResponse.frontend_response:type_name -> temporal.api.workflowservice.v1.DescribeStreamNotifierResponse
+	18, // 8: temporal.server.chasm.lib.streamnotifier.proto.v1.PollStreamNotifierRequest.frontend_request:type_name -> temporal.api.workflowservice.v1.PollStreamNotifierRequest
+	19, // 9: temporal.server.chasm.lib.streamnotifier.proto.v1.PollStreamNotifierResponse.frontend_response:type_name -> temporal.api.workflowservice.v1.PollStreamNotifierResponse
+	10, // [10:10] is the sub-list for method output_type
+	10, // [10:10] is the sub-list for method input_type
+	10, // [10:10] is the sub-list for extension type_name
+	10, // [10:10] is the sub-list for extension extendee
+	0,  // [0:10] is the sub-list for field type_name
 }
 
 func init() { file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_init() }
@@ -531,7 +648,7 @@ func file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_pro
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_rawDesc), len(file_temporal_server_chasm_lib_streamnotifier_proto_v1_request_response_proto_rawDesc)),
 			NumEnums:      0,
-			NumMessages:   8,
+			NumMessages:   10,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
diff --git a/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/service.pb.go b/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/service.pb.go
index 5dfe956de2a..bf129307cfa 100644
--- a/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/service.pb.go
+++ b/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/service.pb.go
@@ -27,34 +27,39 @@ var File_temporal_server_chasm_lib_streamnotifier_proto_v1_service_proto protore
 
 const file_temporal_server_chasm_lib_streamnotifier_proto_v1_service_proto_rawDesc = "" +
 	"\n" +
-	"?temporal/server/chasm/lib/streamnotifier/proto/v1/service.proto\x121temporal.server.chasm.lib.streamnotifier.proto.v1\x1aHtemporal/server/chasm/lib/streamnotifier/proto/v1/request_response.proto\x1a0temporal/server/api/common/v1/api_category.proto\x1a.temporal/server/api/routing/v1/extension.proto2\xd1\x06\n" +
+	"?temporal/server/chasm/lib/streamnotifier/proto/v1/service.proto\x121temporal.server.chasm.lib.streamnotifier.proto.v1\x1aHtemporal/server/chasm/lib/streamnotifier/proto/v1/request_response.proto\x1a0temporal/server/api/common/v1/api_category.proto\x1a.temporal/server/api/routing/v1/extension.proto2\x9e\b\n" +
 	"\x15StreamNotifierService\x12\xd0\x01\n" +
 	"\x14AttachStreamCallback\x12N.temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackRequest\x1aO.temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackResponse\"\x17\x8a\xb5\x18\x02\b\x01\xd2\xc3\x18\r\x1a\vbusiness_id\x12\xd0\x01\n" +
 	"\x14DetachStreamCallback\x12N.temporal.server.chasm.lib.streamnotifier.proto.v1.DetachStreamCallbackRequest\x1aO.temporal.server.chasm.lib.streamnotifier.proto.v1.DetachStreamCallbackResponse\"\x17\x8a\xb5\x18\x02\b\x01\xd2\xc3\x18\r\x1a\vbusiness_id\x12\xb8\x01\n" +
 	"\fNotifyStream\x12F.temporal.server.chasm.lib.streamnotifier.proto.v1.NotifyStreamRequest\x1aG.temporal.server.chasm.lib.streamnotifier.proto.v1.NotifyStreamResponse\"\x17\x8a\xb5\x18\x02\b\x01\xd2\xc3\x18\r\x1a\vbusiness_id\x12\xd6\x01\n" +
-	"\x16DescribeStreamNotifier\x12P.temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierRequest\x1aQ.temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierResponse\"\x17\x8a\xb5\x18\x02\b\x01\xd2\xc3\x18\r\x1a\vbusiness_idBVZTgo.temporal.io/server/chasm/lib/streamnotifier/gen/streamnotifierpb;streamnotifierpbb\x06proto3"
+	"\x16DescribeStreamNotifier\x12P.temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierRequest\x1aQ.temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierResponse\"\x17\x8a\xb5\x18\x02\b\x01\xd2\xc3\x18\r\x1a\vbusiness_id\x12\xca\x01\n" +
+	"\x12PollStreamNotifier\x12L.temporal.server.chasm.lib.streamnotifier.proto.v1.PollStreamNotifierRequest\x1aM.temporal.server.chasm.lib.streamnotifier.proto.v1.PollStreamNotifierResponse\"\x17\x8a\xb5\x18\x02\b\x02\xd2\xc3\x18\r\x1a\vbusiness_idBVZTgo.temporal.io/server/chasm/lib/streamnotifier/gen/streamnotifierpb;streamnotifierpbb\x06proto3"
 
 var file_temporal_server_chasm_lib_streamnotifier_proto_v1_service_proto_goTypes = []any{
 	(*AttachStreamCallbackRequest)(nil),    // 0: temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackRequest
 	(*DetachStreamCallbackRequest)(nil),    // 1: temporal.server.chasm.lib.streamnotifier.proto.v1.DetachStreamCallbackRequest
 	(*NotifyStreamRequest)(nil),            // 2: temporal.server.chasm.lib.streamnotifier.proto.v1.NotifyStreamRequest
 	(*DescribeStreamNotifierRequest)(nil),  // 3: temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierRequest
-	(*AttachStreamCallbackResponse)(nil),   // 4: temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackResponse
-	(*DetachStreamCallbackResponse)(nil),   // 5: temporal.server.chasm.lib.streamnotifier.proto.v1.DetachStreamCallbackResponse
-	(*NotifyStreamResponse)(nil),           // 6: temporal.server.chasm.lib.streamnotifier.proto.v1.NotifyStreamResponse
-	(*DescribeStreamNotifierResponse)(nil), // 7: temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierResponse
+	(*PollStreamNotifierRequest)(nil),      // 4: temporal.server.chasm.lib.streamnotifier.proto.v1.PollStreamNotifierRequest
+	(*AttachStreamCallbackResponse)(nil),   // 5: temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackResponse
+	(*DetachStreamCallbackResponse)(nil),   // 6: temporal.server.chasm.lib.streamnotifier.proto.v1.DetachStreamCallbackResponse
+	(*NotifyStreamResponse)(nil),           // 7: temporal.server.chasm.lib.streamnotifier.proto.v1.NotifyStreamResponse
+	(*DescribeStreamNotifierResponse)(nil), // 8: temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierResponse
+	(*PollStreamNotifierResponse)(nil),     // 9: temporal.server.chasm.lib.streamnotifier.proto.v1.PollStreamNotifierResponse
 }
 var file_temporal_server_chasm_lib_streamnotifier_proto_v1_service_proto_depIdxs = []int32{
 	0, // 0: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.AttachStreamCallback:input_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackRequest
 	1, // 1: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.DetachStreamCallback:input_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.DetachStreamCallbackRequest
 	2, // 2: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.NotifyStream:input_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.NotifyStreamRequest
 	3, // 3: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.DescribeStreamNotifier:input_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierRequest
-	4, // 4: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.AttachStreamCallback:output_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackResponse
-	5, // 5: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.DetachStreamCallback:output_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.DetachStreamCallbackResponse
-	6, // 6: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.NotifyStream:output_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.NotifyStreamResponse
-	7, // 7: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.DescribeStreamNotifier:output_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierResponse
-	4, // [4:8] is the sub-list for method output_type
-	0, // [0:4] is the sub-list for method input_type
+	4, // 4: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.PollStreamNotifier:input_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.PollStreamNotifierRequest
+	5, // 5: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.AttachStreamCallback:output_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.AttachStreamCallbackResponse
+	6, // 6: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.DetachStreamCallback:output_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.DetachStreamCallbackResponse
+	7, // 7: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.NotifyStream:output_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.NotifyStreamResponse
+	8, // 8: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.DescribeStreamNotifier:output_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.DescribeStreamNotifierResponse
+	9, // 9: temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService.PollStreamNotifier:output_type -> temporal.server.chasm.lib.streamnotifier.proto.v1.PollStreamNotifierResponse
+	5, // [5:10] is the sub-list for method output_type
+	0, // [0:5] is the sub-list for method input_type
 	0, // [0:0] is the sub-list for extension type_name
 	0, // [0:0] is the sub-list for extension extendee
 	0, // [0:0] is the sub-list for field type_name
diff --git a/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/service_client.pb.go b/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/service_client.pb.go
index bda1e1700a6..80ade95ad1b 100644
--- a/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/service_client.pb.go
+++ b/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/service_client.pb.go
@@ -237,3 +237,46 @@ func (c *StreamNotifierServiceLayeredClient) DescribeStreamNotifier(
 	}
 	return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
 }
+func (c *StreamNotifierServiceLayeredClient) callPollStreamNotifierNoRetry(
+	ctx context.Context,
+	request *PollStreamNotifierRequest,
+	opts ...grpc.CallOption,
+) (*PollStreamNotifierResponse, error) {
+	var response *PollStreamNotifierResponse
+	var err error
+	startTime := time.Now().UTC()
+	// the caller is a namespace, hence the tag below.
+	caller := headers.GetCallerInfo(ctx).CallerName
+	metricsHandler := c.metricsHandler.WithTags(
+		metrics.OperationTag("StreamNotifierService.PollStreamNotifier"),
+		metrics.NamespaceTag(caller),
+		metrics.ServiceRoleTag(metrics.HistoryRoleTagValue),
+	)
+	metrics.ClientRequests.With(metricsHandler).Record(1)
+	defer func() {
+		if err != nil {
+			metrics.ClientFailures.With(metricsHandler).Record(1, metrics.ServiceErrorTypeTag(err))
+		}
+		metrics.ClientLatency.With(metricsHandler).Record(time.Since(startTime))
+	}()
+	shardID := common.WorkflowIDToHistoryShard(request.GetNamespaceId(), request.GetBusinessId(), c.numShards)
+	op := func(ctx context.Context, client StreamNotifierServiceClient) error {
+		var err error
+		ctx, cancel := context.WithTimeout(ctx, history.DefaultTimeout)
+		defer cancel()
+		response, err = client.PollStreamNotifier(ctx, request, opts...)
+		return err
+	}
+	err = c.redirector.Execute(ctx, shardID, op)
+	return response, err
+}
+func (c *StreamNotifierServiceLayeredClient) PollStreamNotifier(
+	ctx context.Context,
+	request *PollStreamNotifierRequest,
+	opts ...grpc.CallOption,
+) (*PollStreamNotifierResponse, error) {
+	call := func(ctx context.Context) (*PollStreamNotifierResponse, error) {
+		return c.callPollStreamNotifierNoRetry(ctx, request, opts...)
+	}
+	return backoff.ThrottleRetryContextWithReturn(ctx, call, c.retryPolicy, common.IsServiceClientTransientError)
+}
diff --git a/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/service_grpc.pb.go b/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/service_grpc.pb.go
index 264600b53f6..87a20090f65 100644
--- a/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/service_grpc.pb.go
+++ b/chasm/lib/streamnotifier/gen/streamnotifierpb/v1/service_grpc.pb.go
@@ -24,6 +24,7 @@ const (
 	StreamNotifierService_DetachStreamCallback_FullMethodName   = "/temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService/DetachStreamCallback"
 	StreamNotifierService_NotifyStream_FullMethodName           = "/temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService/NotifyStream"
 	StreamNotifierService_DescribeStreamNotifier_FullMethodName = "/temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService/DescribeStreamNotifier"
+	StreamNotifierService_PollStreamNotifier_FullMethodName     = "/temporal.server.chasm.lib.streamnotifier.proto.v1.StreamNotifierService/PollStreamNotifier"
 )
 
 // StreamNotifierServiceClient is the client API for StreamNotifierService service.
@@ -34,6 +35,7 @@ type StreamNotifierServiceClient interface {
 	DetachStreamCallback(ctx context.Context, in *DetachStreamCallbackRequest, opts ...grpc.CallOption) (*DetachStreamCallbackResponse, error)
 	NotifyStream(ctx context.Context, in *NotifyStreamRequest, opts ...grpc.CallOption) (*NotifyStreamResponse, error)
 	DescribeStreamNotifier(ctx context.Context, in *DescribeStreamNotifierRequest, opts ...grpc.CallOption) (*DescribeStreamNotifierResponse, error)
+	PollStreamNotifier(ctx context.Context, in *PollStreamNotifierRequest, opts ...grpc.CallOption) (*PollStreamNotifierResponse, error)
 }
 
 type streamNotifierServiceClient struct {
@@ -80,6 +82,15 @@ func (c *streamNotifierServiceClient) DescribeStreamNotifier(ctx context.Context
 	return out, nil
 }
 
+func (c *streamNotifierServiceClient) PollStreamNotifier(ctx context.Context, in *PollStreamNotifierRequest, opts ...grpc.CallOption) (*PollStreamNotifierResponse, error) {
+	out := new(PollStreamNotifierResponse)
+	err := c.cc.Invoke(ctx, StreamNotifierService_PollStreamNotifier_FullMethodName, in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
 // StreamNotifierServiceServer is the server API for StreamNotifierService service.
 // All implementations must embed UnimplementedStreamNotifierServiceServer
 // for forward compatibility
@@ -88,6 +99,7 @@ type StreamNotifierServiceServer interface {
 	DetachStreamCallback(context.Context, *DetachStreamCallbackRequest) (*DetachStreamCallbackResponse, error)
 	NotifyStream(context.Context, *NotifyStreamRequest) (*NotifyStreamResponse, error)
 	DescribeStreamNotifier(context.Context, *DescribeStreamNotifierRequest) (*DescribeStreamNotifierResponse, error)
+	PollStreamNotifier(context.Context, *PollStreamNotifierRequest) (*PollStreamNotifierResponse, error)
 	mustEmbedUnimplementedStreamNotifierServiceServer()
 }
 
@@ -107,6 +119,9 @@ func (UnimplementedStreamNotifierServiceServer) NotifyStream(context.Context, *N
 func (UnimplementedStreamNotifierServiceServer) DescribeStreamNotifier(context.Context, *DescribeStreamNotifierRequest) (*DescribeStreamNotifierResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method DescribeStreamNotifier not implemented")
 }
+func (UnimplementedStreamNotifierServiceServer) PollStreamNotifier(context.Context, *PollStreamNotifierRequest) (*PollStreamNotifierResponse, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method PollStreamNotifier not implemented")
+}
 func (UnimplementedStreamNotifierServiceServer) mustEmbedUnimplementedStreamNotifierServiceServer() {}
 
 // UnsafeStreamNotifierServiceServer may be embedded to opt out of forward compatibility for this service.
@@ -192,6 +207,24 @@ func _StreamNotifierService_DescribeStreamNotifier_Handler(srv interface{}, ctx
 	return interceptor(ctx, in, info, handler)
 }
 
+func _StreamNotifierService_PollStreamNotifier_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(PollStreamNotifierRequest)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(StreamNotifierServiceServer).PollStreamNotifier(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: StreamNotifierService_PollStreamNotifier_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(StreamNotifierServiceServer).PollStreamNotifier(ctx, req.(*PollStreamNotifierRequest))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
 // StreamNotifierService_ServiceDesc is the grpc.ServiceDesc for StreamNotifierService service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -215,6 +248,10 @@ var StreamNotifierService_ServiceDesc = grpc.ServiceDesc{
 			MethodName: "DescribeStreamNotifier",
 			Handler:    _StreamNotifierService_DescribeStreamNotifier_Handler,
 		},
+		{
+			MethodName: "PollStreamNotifier",
+			Handler:    _StreamNotifierService_PollStreamNotifier_Handler,
+		},
 	},
 	Streams:  []grpc.StreamDesc{},
 	Metadata: "temporal/server/chasm/lib/streamnotifier/proto/v1/service.proto",
diff --git a/chasm/lib/streamnotifier/handler.go b/chasm/lib/streamnotifier/handler.go
index 7de3daec5ad..d3c3957dc85 100644
--- a/chasm/lib/streamnotifier/handler.go
+++ b/chasm/lib/streamnotifier/handler.go
@@ -12,6 +12,7 @@ import (
 	"go.temporal.io/server/chasm"
 	streamnotifierpb "go.temporal.io/server/chasm/lib/streamnotifier/gen/streamnotifierpb/v1"
 	"go.temporal.io/server/common"
+	"go.temporal.io/server/common/contextutil"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/resource"
 )
@@ -126,28 +127,18 @@ func (h *handler) NotifyStream(
 			closeFailure: r.GetCloseFailure(),
 			idleTimeout:  h.config.IdleTimeout(ns),
 			retention:    h.config.ClosedRetention(ns),
+
+			idleTimeoutNoCallbacks: h.config.IdleTimeoutNoCallbacks(ns),
 		})
 	}
-	if fe.GetClose() {
-		// A late attach must find the stream closed, so a close starts a notifier if none runs.
-		if err := startOrUpdate(
-			ctx, req.GetNamespaceId(), req.GetBusinessId(), fe.GetStreamRef(), update, fe,
-		); err != nil {
-			return nil, err
-		}
-		return &streamnotifierpb.NotifyStreamResponse{
-			FrontendResponse: &workflowservice.NotifyStreamResponse{},
-		}, nil
-	}
-	// No notifier means no caller waits, and a caller that attaches later reads from its start
-	// position. So a notification starts no notifier, and one that names the wrong run of the
-	// owner's chain leaves nothing behind.
-	ref := chasm.NewComponentRef[*StreamNotifier](chasm.ExecutionKey{
-		NamespaceID: req.GetNamespaceId(),
-		BusinessID:  req.GetBusinessId(),
-	})
-	_, _, err := chasm.UpdateComponent(ctx, ref, update, fe)
-	if _, ok := errors.AsType[*serviceerror.NotFound](err); ok {
+	// A notification starts a notifier when the stream has none. A reader that armed the store
+	// may not have reached the notifier yet, and the notifier keeps the counter it waits for. A
+	// late attach must also find a closed stream closed.
+	err := startOrUpdate(
+		ctx, req.GetNamespaceId(), req.GetBusinessId(), fe.GetStreamRef(), update, fe,
+	)
+	if _, ok := errors.AsType[*serviceerror.FailedPrecondition](err); ok && !fe.GetClose() {
+		// The stream closed and its notifier ended. Nothing waits on a notification.
 		err = nil
 	}
 	if err != nil {
@@ -224,6 +215,103 @@ func (h *handler) DescribeStreamNotifier(
 	return &streamnotifierpb.DescribeStreamNotifierResponse{FrontendResponse: resp}, nil
 }
 
+// PollStreamNotifier waits until the notifier's counter passes the request's after_counter, the
+// stream closes, or the wait limit passes. A poll writes nothing, except that it starts a notifier
+// when the stream has none, so the notification the client waits for has somewhere to land.
+func (h *handler) PollStreamNotifier(
+	ctx context.Context,
+	req *streamnotifierpb.PollStreamNotifierRequest,
+) (_ *streamnotifierpb.PollStreamNotifierResponse, retErr error) {
+	defer log.CapturePanic(h.logger, &retErr)
+	fe := req.GetFrontendRequest()
+	ns := fe.GetNamespace()
+	after := fe.GetAfterCounter()
+	ref := chasm.NewComponentRef[*StreamNotifier](chasm.ExecutionKey{
+		NamespaceID: req.GetNamespaceId(),
+		BusinessID:  req.GetBusinessId(),
+	})
+	type snapshot struct {
+		response *workflowservice.PollStreamNotifierResponse
+		done     bool
+		ended    bool
+	}
+	current, err := chasm.ReadComponent(
+		ctx,
+		ref,
+		func(n *StreamNotifier, cctx chasm.Context, _ struct{}) (snapshot, error) {
+			resp, err := n.pollResponse(cctx, struct{}{})
+			return snapshot{
+				response: resp,
+				done:     n.pollDone(after),
+				ended:    n.Expired && !n.Closed,
+			}, err
+		},
+		struct{}{},
+	)
+	if _, ok := errors.AsType[*serviceerror.NotFound](err); ok {
+		current, err = snapshot{ended: true}, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	if current.ended {
+		// No notifier, or one that ended idle: start a new one and wait on it.
+		idle := h.config.IdleTimeoutNoCallbacks(ns)
+		if err := startOrUpdate(
+			ctx,
+			req.GetNamespaceId(),
+			req.GetBusinessId(),
+			fe.GetStreamRef(),
+			func(n *StreamNotifier, mctx chasm.MutableContext, _ struct{}) (struct{}, error) {
+				n.startIdle(mctx, idle)
+				return struct{}{}, nil
+			},
+			struct{}{},
+		); err != nil {
+			return nil, err
+		}
+		current = snapshot{response: &workflowservice.PollStreamNotifierResponse{}}
+	} else if current.done {
+		return &streamnotifierpb.PollStreamNotifierResponse{
+			FrontendResponse: current.response,
+		}, nil
+	}
+
+	pollCtx, cancel := contextutil.WithDeadlineBuffer(
+		ctx,
+		h.config.PollMaxWait(ns),
+		common.DefaultLongPollBuffer,
+	)
+	defer cancel()
+	resp, _, err := chasm.PollComponent(
+		pollCtx,
+		ref,
+		func(
+			n *StreamNotifier,
+			cctx chasm.Context,
+			after int64,
+		) (*workflowservice.PollStreamNotifierResponse, bool, error) {
+			if !n.pollDone(after) {
+				return nil, false, nil
+			}
+			resp, err := n.pollResponse(cctx, struct{}{})
+			return resp, true, err
+		},
+		after,
+	)
+	if err != nil && pollCtx.Err() != nil && ctx.Err() == nil {
+		// The wait limit passed. The client reads the stream again.
+		err, resp = nil, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	if resp == nil {
+		resp = current.response
+	}
+	return &streamnotifierpb.PollStreamNotifierResponse{FrontendResponse: resp}, nil
+}
+
 // checkFirstRun refuses a reference whose run ID is not the first run of its owner's chain. Such a
 // reference would key a second notifier for the stream, and the owner check would take the owner
 // for ended and close that notifier while the stream runs. An owner that does not exist is left to
diff --git a/chasm/lib/streamnotifier/notifier.go b/chasm/lib/streamnotifier/notifier.go
index 6293d1ee71c..9de312560ab 100644
--- a/chasm/lib/streamnotifier/notifier.go
+++ b/chasm/lib/streamnotifier/notifier.go
@@ -13,6 +13,7 @@ import (
 	failurepb "go.temporal.io/api/failure/v1"
 	"go.temporal.io/api/serviceerror"
 	streampb "go.temporal.io/api/stream/v1"
+	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/callback"
 	callbackspb "go.temporal.io/server/chasm/lib/callback/gen/callbackpb/v1"
@@ -333,7 +334,9 @@ type notifyInput struct {
 	// Set instead of closeResult when the stream closes with a failure.
 	closeFailure *failurepb.Failure
 	idleTimeout  time.Duration
-	retention    time.Duration
+	// The idle timeout when no callback waits, such as for a notifier only clients poll.
+	idleTimeoutNoCallbacks time.Duration
+	retention              time.Duration
 }
 
 // notify keeps the highest counter and hands it to every callback as progress. A close completes
@@ -361,10 +364,43 @@ func (n *StreamNotifier) notify(ctx chasm.MutableContext, in notifyInput) error
 		n.touch(ctx, in.retention)
 		return nil
 	}
-	n.touch(ctx, in.idleTimeout)
+	if n.hasWaitingCallbacks(ctx) {
+		n.touch(ctx, in.idleTimeout)
+	} else {
+		n.touch(ctx, in.idleTimeoutNoCallbacks)
+	}
 	return nil
 }
 
+// startIdle arms the expiry of a notifier that a poll started, and changes nothing on one that
+// already saw activity.
+func (n *StreamNotifier) startIdle(ctx chasm.MutableContext, idleTimeout time.Duration) {
+	if n.LastActivityTime == nil {
+		n.touch(ctx, idleTimeout)
+	}
+}
+
+// pollResponse is what a poll answers with: the latest notification and the close outcome.
+func (n *StreamNotifier) pollResponse(
+	_ chasm.Context,
+	_ struct{},
+) (*workflowservice.PollStreamNotifierResponse, error) {
+	return &workflowservice.PollStreamNotifierResponse{
+		Counter:      n.Counter,
+		Position:     n.Position,
+		Metadata:     maps.Clone(n.Metadata),
+		Closed:       n.Closed,
+		CloseResult:  common.CloneProto(n.CloseResult),
+		CloseFailure: common.CloneProto(n.CloseFailure),
+	}, nil
+}
+
+// pollDone reports whether a poll that waits for a counter above after returns: the counter
+// passed it, the stream closed, or the notifier ended, so the client starts a new one.
+func (n *StreamNotifier) pollDone(after int64) bool {
+	return n.Counter > after || n.Closed || n.Expired
+}
+
 // close completes every callback still waiting: with the result, or with failure when it is set.
 func (n *StreamNotifier) close(
 	ctx chasm.MutableContext,
diff --git a/chasm/lib/streamnotifier/notifier_test.go b/chasm/lib/streamnotifier/notifier_test.go
index 1409dcaffba..c8b348a2ec5 100644
--- a/chasm/lib/streamnotifier/notifier_test.go
+++ b/chasm/lib/streamnotifier/notifier_test.go
@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/nexus-rpc/sdk-go/nexus"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
@@ -28,6 +29,7 @@ import (
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/namespace"
 	commonnexus "go.temporal.io/server/common/nexus"
+	"go.temporal.io/server/service/history/tasks"
 	"go.uber.org/mock/gomock"
 	"google.golang.org/grpc"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -36,6 +38,7 @@ import (
 type notifierTest struct {
 	t          *testing.T
 	ctx        context.Context
+	engine     *chasmtest.Engine
 	handler    *handler
 	expiry     *expiryTaskHandler
 	ownerCheck *ownerCheckTaskHandler
@@ -50,10 +53,14 @@ type notifierTest struct {
 
 func newNotifierTest(t *testing.T, maxCallbacks int) *notifierTest {
 	config := &Config{
-		Enabled:            dynamicconfig.GetBoolPropertyFnFilteredByNamespace(true),
-		MaxCallbacks:       dynamicconfig.GetIntPropertyFnFilteredByNamespace(maxCallbacks),
-		MaxCallbackBytes:   MaxCallbackBytes.Get(dynamicconfig.NewNoopCollection()),
-		IdleTimeout:        dynamicconfig.GetDurationPropertyFnFilteredByNamespace(time.Hour),
+		Enabled:          dynamicconfig.GetBoolPropertyFnFilteredByNamespace(true),
+		MaxCallbacks:     dynamicconfig.GetIntPropertyFnFilteredByNamespace(maxCallbacks),
+		MaxCallbackBytes: MaxCallbackBytes.Get(dynamicconfig.NewNoopCollection()),
+		IdleTimeout:      dynamicconfig.GetDurationPropertyFnFilteredByNamespace(time.Hour),
+		IdleTimeoutNoCallbacks: dynamicconfig.GetDurationPropertyFnFilteredByNamespace(
+			10 * time.Minute,
+		),
+		PollMaxWait:        dynamicconfig.GetDurationPropertyFnFilteredByNamespace(time.Second),
 		ClosedRetention:    dynamicconfig.GetDurationPropertyFnFilteredByNamespace(time.Minute),
 		OwnerCheckInterval: dynamicconfig.GetDurationPropertyFnFilteredByNamespace(5 * time.Minute),
 		OwnerCloseGrace:    dynamicconfig.GetDurationPropertyFnFilteredByNamespace(time.Minute),
@@ -94,9 +101,11 @@ func newNotifierTest(t *testing.T, maxCallbacks int) *notifierTest {
 	require.NoError(t, registry.Register(&chasm.CoreLibrary{}))
 	require.NoError(t, registry.Register(callback.NewNilLibrary()))
 	require.NoError(t, registry.Register(newLibrary(h, expiry, ownerCheck)))
+	engine := chasmtest.NewEngine(t, registry)
 	*nt = notifierTest{
 		t:          t,
-		ctx:        chasm.NewEngineContext(context.Background(), chasmtest.NewEngine(t, registry)),
+		ctx:        chasm.NewEngineContext(context.Background(), engine),
+		engine:     engine,
 		handler:    h,
 		expiry:     expiry,
 		ownerCheck: ownerCheck,
@@ -257,6 +266,18 @@ func (nt *notifierTest) expire() {
 	require.NoError(nt.t, err)
 }
 
+// hasTimer reports whether the notifier scheduled a timer task for at.
+func (nt *notifierTest) hasTimer(at time.Time) bool {
+	byCategory, err := nt.engine.Tasks(chasm.NewComponentRef[*StreamNotifier](nt.key()))
+	require.NoError(nt.t, err)
+	for _, task := range byCategory[tasks.CategoryTimer] {
+		if task.GetVisibilityTime().Equal(at) {
+			return true
+		}
+	}
+	return false
+}
+
 // updateCallback changes one callback's state, standing in for its delivery tasks.
 func (nt *notifierTest) updateCallback(requestID string, fn func(*callback.Callback)) {
 	_, _, err := chasm.UpdateComponent(nt.ctx, chasm.NewComponentRef[*StreamNotifier](nt.key()),
@@ -562,31 +583,31 @@ func TestStreamNotifier(t *testing.T) {
 		require.Contains(t, nt.callbacks(), "later")
 	})
 
-	t.Run("ANotificationWithoutACallerStartsNoNotifier", func(t *testing.T) {
+	t.Run("ANotificationStartsAMissingNotifier", func(t *testing.T) {
 		nt := newNotifierTest(t, 10)
 		require.NoError(t, nt.notify(1, false))
-		_, err := nt.handler.DescribeStreamNotifier(
-			nt.ctx,
-			&streamnotifierpb.DescribeStreamNotifierRequest{
-				NamespaceId: "namespace-id",
-				BusinessId:  BusinessID(nt.ref),
-				FrontendRequest: &workflowservice.DescribeStreamNotifierRequest{
-					Namespace: "ns",
-					StreamRef: nt.ref,
-				},
-			},
-		)
-		var notFound *serviceerror.NotFound
-		require.ErrorAs(t, err, &notFound, "no caller waits, so nothing needs the notification")
+		desc := nt.describe()
+		require.Equal(t, int64(1), desc.GetCounter(),
+			"the notifier keeps the counter a reader that armed waits for")
+		require.Empty(t, desc.GetCallbacks())
+		require.True(t, nt.hasTimer(nt.now().Add(10*time.Minute)),
+			"a notifier with no callbacks ends after the short idle timeout")
 
+		nt = newNotifierTest(t, 10)
 		require.NoError(t, nt.attach("a"))
 		require.NoError(t, nt.notify(2, false))
-		require.Equal(
-			t,
-			int64(2),
-			nt.describe().GetCounter(),
-			"a notifier a caller started takes it",
-		)
+		require.Equal(t, int64(2), nt.describe().GetCounter())
+		require.True(t, nt.hasTimer(nt.now().Add(time.Hour)),
+			"a notifier with a waiting callback keeps the full idle timeout")
+		require.False(t, nt.hasTimer(nt.now().Add(10*time.Minute)))
+	})
+
+	t.Run("ANotificationAfterTheStreamEndedChangesNothing", func(t *testing.T) {
+		nt := newNotifierTest(t, 10)
+		require.NoError(t, nt.notify(1, true))
+		nt.expire()
+		require.NoError(t, nt.notify(2, false), "nothing waits on a closed stream")
+		require.True(t, nt.describe().GetClosed())
 	})
 
 	t.Run("AClosedStreamStaysClosed", func(t *testing.T) {
@@ -938,3 +959,95 @@ func TestLargeCallbacksHitTheByteLimitBeforeTheCountLimit(t *testing.T) {
 	require.NotContains(t, nt.callbacks(), "caller-0")
 	require.Contains(t, nt.callbacks(), "late")
 }
+
+func (nt *notifierTest) poll(after int64) (*workflowservice.PollStreamNotifierResponse, error) {
+	resp, err := nt.handler.PollStreamNotifier(nt.ctx, &streamnotifierpb.PollStreamNotifierRequest{
+		NamespaceId: "namespace-id",
+		BusinessId:  BusinessID(nt.ref),
+		FrontendRequest: &workflowservice.PollStreamNotifierRequest{
+			Namespace:    "ns",
+			StreamRef:    nt.ref,
+			AfterCounter: after,
+		},
+	})
+	return resp.GetFrontendResponse(), err
+}
+
+func TestPollStreamNotifier(t *testing.T) {
+	t.Run("ACounterAlreadyAboveReturnsAtOnce", func(t *testing.T) {
+		nt := newNotifierTest(t, 10)
+		require.NoError(t, nt.notify(42, false))
+		started := time.Now()
+		resp, err := nt.poll(41)
+		require.NoError(t, err)
+		require.Equal(t, int64(42), resp.GetCounter(),
+			"a notification that landed before the poll is not lost")
+		require.Equal(t, "p", resp.GetPosition())
+		require.Equal(t, map[string]string{"k": "v"}, resp.GetMetadata())
+		require.Less(t, time.Since(started), 500*time.Millisecond)
+	})
+
+	t.Run("ANotificationWakesAWaitingPoll", func(t *testing.T) {
+		nt := newNotifierTest(t, 10)
+		require.NoError(t, nt.notify(1, false))
+		done := make(chan *workflowservice.PollStreamNotifierResponse, 1)
+		go func() {
+			resp, err := nt.poll(1)
+			assert.NoError(t, err)
+			done <- resp
+		}()
+		require.NoError(t, nt.notify(2, false))
+		resp := <-done
+		require.Equal(t, int64(2), resp.GetCounter())
+	})
+
+	t.Run("TheWaitLimitReturnsNoChange", func(t *testing.T) {
+		nt := newNotifierTest(t, 10)
+		require.NoError(t, nt.notify(5, false))
+		resp, err := nt.poll(5)
+		require.NoError(t, err)
+		require.Equal(t, int64(5), resp.GetCounter(), "the counter is not above the one asked for")
+		require.False(t, resp.GetClosed())
+	})
+
+	t.Run("APollOnAMissingNotifierStartsIt", func(t *testing.T) {
+		nt := newNotifierTest(t, 10)
+		resp, err := nt.poll(41)
+		require.NoError(t, err)
+		require.Zero(t, resp.GetCounter())
+		require.Zero(t, nt.describe().GetCounter(), "the poll started the notifier")
+		require.True(t, nt.hasTimer(nt.now().Add(10*time.Minute)),
+			"it ends after the short idle timeout if nothing happens")
+	})
+
+	t.Run("APollOnAnEndedNotifierStartsANewOne", func(t *testing.T) {
+		nt := newNotifierTest(t, 10)
+		require.NoError(t, nt.notify(1, false))
+		nt.expire()
+		nt.read(func(n *StreamNotifier, ctx chasm.Context) {
+			require.Equal(t, chasm.LifecycleStateFailed, n.LifecycleState(ctx))
+		})
+		resp, err := nt.poll(1)
+		require.NoError(t, err)
+		require.Zero(t, resp.GetCounter(), "the poll waited on a new notifier")
+		nt.read(func(n *StreamNotifier, ctx chasm.Context) {
+			require.Equal(t, chasm.LifecycleStateRunning, n.LifecycleState(ctx))
+		})
+	})
+
+	t.Run("TheCloseEndsThePollWithItsOutcome", func(t *testing.T) {
+		nt := newNotifierTest(t, 10)
+		require.NoError(t, nt.notify(3, true))
+		resp, err := nt.poll(100)
+		require.NoError(t, err)
+		require.True(t, resp.GetClosed())
+		require.Equal(t, []byte("summary"), resp.GetCloseResult().GetData())
+
+		nt = newNotifierTest(t, 10)
+		require.NoError(t, nt.closeWithFailure(streamFailure("boom")))
+		resp, err = nt.poll(100)
+		require.NoError(t, err)
+		require.True(t, resp.GetClosed())
+		require.Equal(t, "boom", resp.GetCloseFailure().GetMessage())
+	})
+}
diff --git a/chasm/lib/streamnotifier/proto/v1/request_response.proto b/chasm/lib/streamnotifier/proto/v1/request_response.proto
index 15859ad806b..8213bfb863e 100644
--- a/chasm/lib/streamnotifier/proto/v1/request_response.proto
+++ b/chasm/lib/streamnotifier/proto/v1/request_response.proto
@@ -48,3 +48,13 @@ message DescribeStreamNotifierRequest {
 message DescribeStreamNotifierResponse {
   temporal.api.workflowservice.v1.DescribeStreamNotifierResponse frontend_response = 1;
 }
+
+message PollStreamNotifierRequest {
+  string namespace_id = 1;
+  string business_id = 2;
+  temporal.api.workflowservice.v1.PollStreamNotifierRequest frontend_request = 3;
+}
+
+message PollStreamNotifierResponse {
+  temporal.api.workflowservice.v1.PollStreamNotifierResponse frontend_response = 1;
+}
diff --git a/chasm/lib/streamnotifier/proto/v1/service.proto b/chasm/lib/streamnotifier/proto/v1/service.proto
index cf9b2c985ac..6e3824bfb0f 100644
--- a/chasm/lib/streamnotifier/proto/v1/service.proto
+++ b/chasm/lib/streamnotifier/proto/v1/service.proto
@@ -28,4 +28,9 @@ service StreamNotifierService {
     option (temporal.server.api.routing.v1.routing).business_id = "business_id";
     option (temporal.server.api.common.v1.api_category).category = API_CATEGORY_STANDARD;
   }
+
+  rpc PollStreamNotifier(PollStreamNotifierRequest) returns (PollStreamNotifierResponse) {
+    option (temporal.server.api.routing.v1.routing).business_id = "business_id";
+    option (temporal.server.api.common.v1.api_category).category = API_CATEGORY_LONG_POLL;
+  }
 }
diff --git a/common/rpc/interceptor/dc_redirection_policy.go b/common/rpc/interceptor/dc_redirection_policy.go
index 9900dab3d1c..64923febaeb 100644
--- a/common/rpc/interceptor/dc_redirection_policy.go
+++ b/common/rpc/interceptor/dc_redirection_policy.go
@@ -105,6 +105,7 @@ var selectedAPIsForwardingRedirectionPolicyAllowedAPIs = map[string]struct{}{
 	wfMethod("AttachStreamCallback"): {},
 	wfMethod("DetachStreamCallback"): {},
 	wfMethod("NotifyStream"):         {},
+	wfMethod("PollStreamNotifier"):   {},
 }
 
 func wfMethod(name string) string { return api.WorkflowServicePrefix + name }
diff --git a/common/rpc/interceptor/dc_redirection_policy_test.go b/common/rpc/interceptor/dc_redirection_policy_test.go
index a03ac536dbe..0b08f6fd019 100644
--- a/common/rpc/interceptor/dc_redirection_policy_test.go
+++ b/common/rpc/interceptor/dc_redirection_policy_test.go
@@ -163,6 +163,7 @@ func (s *selectedAPIsForwardingRedirectionPolicySuite) TestSelectedAPIs() {
 		wfMethod("AttachStreamCallback"): {},
 		wfMethod("DetachStreamCallback"): {},
 		wfMethod("NotifyStream"):         {},
+		wfMethod("PollStreamNotifier"):   {},
 	}, selectedAPIsForwardingRedirectionPolicyAllowedAPIs)
 }
 
diff --git a/service/frontend/stream_notifier_handler.go b/service/frontend/stream_notifier_handler.go
deleted file mode 100644
index 4ee1ca94aba..00000000000
--- a/service/frontend/stream_notifier_handler.go
+++ /dev/null
@@ -1,16 +0,0 @@
-package frontend
-
-import (
-	"context"
-
-	"go.temporal.io/api/serviceerror"
-	"go.temporal.io/api/workflowservice/v1"
-)
-
-// PollStreamNotifier is declared by the API this server pins but not served yet, so it answers
-// Unimplemented rather than leaving the service without it.
-func (wh *WorkflowHandler) PollStreamNotifier(
-	context.Context, *workflowservice.PollStreamNotifierRequest,
-) (*workflowservice.PollStreamNotifierResponse, error) {
-	return nil, serviceerror.NewUnimplemented("PollStreamNotifier is not implemented")
-}
diff --git a/tests/stream_poll_test.go b/tests/stream_poll_test.go
new file mode 100644
index 00000000000..72affa930af
--- /dev/null
+++ b/tests/stream_poll_test.go
@@ -0,0 +1,136 @@
+package tests
+
+import (
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	streampb "go.temporal.io/api/stream/v1"
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/sdk/converter"
+	"go.temporal.io/server/chasm/lib/streamnotifier"
+	"go.temporal.io/server/common/dynamicconfig"
+	"go.temporal.io/server/common/testing/parallelsuite"
+	"go.temporal.io/server/tests/testcore"
+)
+
+// StreamPollSuite covers PollStreamNotifier, the long poll a client reader waits on after a read
+// that found nothing and armed the store.
+type StreamPollSuite struct {
+	parallelsuite.Suite[*StreamPollSuite]
+}
+
+func TestStreamPollSuite(t *testing.T) {
+	parallelsuite.Run(t, &StreamPollSuite{})
+}
+
+type streamPollEnv struct {
+	*testcore.TestEnv
+	stream *streampb.StreamReference
+}
+
+func (s *StreamPollSuite) newEnv() *streamPollEnv {
+	env := testcore.NewEnv(
+		s.T(),
+		testcore.WithDynamicConfig(dynamicconfig.EnableChasm, true),
+		testcore.WithDynamicConfig(streamnotifier.Enabled, true),
+		testcore.WithDynamicConfig(streamnotifier.PollMaxWait, 2*time.Second),
+	)
+	return &streamPollEnv{
+		TestEnv: env,
+		stream: &streampb.StreamReference{
+			OwnerKind:  enumspb.STREAM_OWNER_KIND_WORKFLOW,
+			WorkflowId: "producer-" + uuid.NewString(),
+			RunId:      "first-run",
+			Topic:      "tokens",
+		},
+	}
+}
+
+func (s *StreamPollSuite) notify(e *streamPollEnv, counter int64, closeResult *commonpb.Payload) {
+	_, err := e.FrontendClient().NotifyStream(s.Context(), &workflowservice.NotifyStreamRequest{
+		Namespace:   e.Namespace().String(),
+		StreamRef:   e.stream,
+		Position:    "cursor",
+		Counter:     counter,
+		Metadata:    map[string]string{"topic": "tokens"},
+		Close:       closeResult != nil,
+		CloseResult: closeResult,
+	})
+	s.NoError(err)
+}
+
+func (s *StreamPollSuite) poll(
+	e *streamPollEnv,
+	after int64,
+) *workflowservice.PollStreamNotifierResponse {
+	resp, err := e.FrontendClient().PollStreamNotifier(
+		s.Context(),
+		&workflowservice.PollStreamNotifierRequest{
+			Namespace:    e.Namespace().String(),
+			StreamRef:    e.stream,
+			AfterCounter: after,
+		},
+	)
+	s.NoError(err)
+	return resp
+}
+
+func (s *StreamPollSuite) TestANotificationBeforeThePollIsNotLost() {
+	env := s.newEnv()
+	// The producer notifies before the reader polls, and no notifier existed yet.
+	s.notify(env, 42, nil)
+	started := time.Now()
+	resp := s.poll(env, 41)
+	s.Equal(int64(42), resp.GetCounter())
+	s.Equal("cursor", resp.GetPosition())
+	s.Equal("tokens", resp.GetMetadata()["topic"])
+	s.Less(time.Since(started), time.Second, "the poll returns at once")
+}
+
+func (s *StreamPollSuite) TestANotificationWakesAWaitingPoll() {
+	env := s.newEnv()
+	s.notify(env, 1, nil)
+	done := make(chan *workflowservice.PollStreamNotifierResponse, 1)
+	go func() { done <- s.poll(env, 1) }()
+	// Either order works: a poll that starts after the notification returns at once.
+	s.notify(env, 2, nil)
+	resp := <-done
+	s.Equal(int64(2), resp.GetCounter())
+}
+
+func (s *StreamPollSuite) TestTheWaitLimitReturnsNoChange() {
+	env := s.newEnv()
+	started := time.Now()
+	resp := s.poll(env, 7)
+	s.LessOrEqual(resp.GetCounter(), int64(7), "the client reads the stream again")
+	s.False(resp.GetClosed())
+	s.GreaterOrEqual(time.Since(started), time.Second, "the poll waited")
+	s.Less(time.Since(started), 10*time.Second, "streamnotifier.pollMaxWait bounds the wait")
+
+	// The poll started the notifier, so a later notification has somewhere to land.
+	desc, err := env.FrontendClient().DescribeStreamNotifier(
+		s.Context(),
+		&workflowservice.DescribeStreamNotifierRequest{
+			Namespace: env.Namespace().String(),
+			StreamRef: env.stream,
+		},
+	)
+	s.NoError(err)
+	s.Zero(desc.GetCounter())
+}
+
+func (s *StreamPollSuite) TestTheCloseEndsThePoll() {
+	env := s.newEnv()
+	s.notify(env, 1, nil)
+	done := make(chan *workflowservice.PollStreamNotifierResponse, 1)
+	go func() { done <- s.poll(env, 100) }()
+	s.notify(env, 2, testcore.MustToPayload(s.T(), "summary"))
+	resp := <-done
+	s.True(resp.GetClosed())
+	var summary string
+	s.NoError(converter.GetDefaultDataConverter().FromPayload(resp.GetCloseResult(), &summary))
+	s.Equal("summary", summary)
+}