From 688438dacb0f6f16248142a4c94deb6fa59e2c9e Mon Sep 17 00:00:00 2001 From: Julius Park Date: Tue, 13 Oct 2026 10:42:27 -0400 Subject: [PATCH 1/6] initial commit --- api-contracts/v1/streams.proto | 8 + .../migrations/20261013133204_v1_0_172.sql | 10 + examples/typescript/durable-streams/run.ts | 90 ++++++++ examples/typescript/durable-streams/worker.ts | 14 ++ .../typescript/durable-streams/workflow.ts | 21 ++ .../docs/reference/changelog/typescript.mdx | 6 + .../typescript/feature-clients/streams.mdx | 22 +- .../services/shared/proto/v1/streams.pb.go | 192 +++++++++++++++--- .../shared/proto/v1/streams_grpc.pb.go | 36 ++++ .../proto/v1/v1connect/streams.connect.go | 32 +++ internal/services/streams/server.go | 48 +++++ pkg/repository/sqlcv1/models.go | 1 + pkg/repository/sqlcv1/streams.sql | 12 ++ pkg/repository/sqlcv1/streams.sql.go | 41 +++- pkg/repository/streams.go | 36 +++- pkg/repository/streams_repository_test.go | 33 +++ sdks/typescript/CHANGELOG.md | 6 + sdks/typescript/package.json | 2 +- .../typescript/src/protoc-es/v1/streams_pb.ts | 45 +++- sdks/typescript/src/protoc/v1/streams.ts | 145 +++++++++++++ .../src/v1/client/features/streams.test.ts | 33 +++ .../src/v1/client/features/streams.ts | 20 +- .../durable-streams/durable-streams.e2e.ts | 18 ++ .../src/v1/examples/durable-streams/run.ts | 96 +++++++++ .../src/v1/examples/durable-streams/worker.ts | 14 ++ .../v1/examples/durable-streams/workflow.ts | 22 ++ sdks/typescript/src/version.ts | 2 +- sql/schema/v1-core.sql | 2 + 28 files changed, 969 insertions(+), 38 deletions(-) create mode 100644 cmd/hatchet-migrate/migrate/migrations/20261013133204_v1_0_172.sql create mode 100644 examples/typescript/durable-streams/run.ts create mode 100644 examples/typescript/durable-streams/worker.ts create mode 100644 examples/typescript/durable-streams/workflow.ts create mode 100644 sdks/typescript/src/v1/examples/durable-streams/run.ts create mode 100644 sdks/typescript/src/v1/examples/durable-streams/worker.ts create mode 100644 sdks/typescript/src/v1/examples/durable-streams/workflow.ts diff --git a/api-contracts/v1/streams.proto b/api-contracts/v1/streams.proto index 7014d5de81..90f550cb38 100644 --- a/api-contracts/v1/streams.proto +++ b/api-contracts/v1/streams.proto @@ -10,6 +10,7 @@ service V1Streams { rpc Publish(PublishStreamMessageRequest) returns (PublishStreamMessageResponse) {} rpc Subscribe(SubscribeStreamRequest) returns (stream StreamMessage) {} rpc GetTopicMetadata(GetStreamTopicMetadataRequest) returns (StreamTopicMetadata) {} + rpc Reset(ResetStreamTopicRequest) returns (ResetStreamTopicResponse) {} } message PublishStreamMessageRequest { @@ -61,3 +62,10 @@ message StreamTopicMetadata { // when the newest retained message was stored; unset when none is retained google.protobuf.Timestamp last_published_at = 6; } + +message ResetStreamTopicRequest { + string namespace = 1; + string topic = 2; +} + +message ResetStreamTopicResponse {} diff --git a/cmd/hatchet-migrate/migrate/migrations/20261013133204_v1_0_172.sql b/cmd/hatchet-migrate/migrate/migrations/20261013133204_v1_0_172.sql new file mode 100644 index 0000000000..fe4601e620 --- /dev/null +++ b/cmd/hatchet-migrate/migrate/migrations/20261013133204_v1_0_172.sql @@ -0,0 +1,10 @@ +-- +goose Up +-- +goose StatementBegin +-- subscribers without a cursor start after this offset (see ResetStreamTopic) +ALTER TABLE v1_stream_topic ADD COLUMN reset_offset BIGINT NOT NULL DEFAULT 0; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +ALTER TABLE v1_stream_topic DROP COLUMN reset_offset; +-- +goose StatementEnd diff --git a/examples/typescript/durable-streams/run.ts b/examples/typescript/durable-streams/run.ts new file mode 100644 index 0000000000..7b4b901523 --- /dev/null +++ b/examples/typescript/durable-streams/run.ts @@ -0,0 +1,90 @@ +import { randomUUID } from 'crypto'; +import { hatchet } from '../hatchet-client'; +import { progressTask } from './workflow'; + +const decode = (payload: Uint8Array) => new TextDecoder().decode(payload); + +export async function consume() { + // > Consume + const topic = `progress-${randomUUID()}`; + await progressTask.runNoWait({ topic }); + + for await (const event of hatchet.streams.events(topic)) { + const { step, total } = JSON.parse(decode(event.payload)); + console.log(`step ${step} of ${total}`); + + // the stream outlives the run, so the reader decides when it's done + if (step === total) { + break; + } + } +} + +export async function publishFromAnywhere() { + // > Publish from anywhere + await hatchet.streams.publish( + 'orders', + JSON.stringify({ orderId: 'ord_123', status: 'shipped' }) + ); +} + +export async function resumeFromCursor(topic: string) { + // > Resume from a cursor + let cursor: string | undefined; + + for await (const event of hatchet.streams.events(topic)) { + console.log(decode(event.payload)); + ({ cursor } = event); + break; + } + + // picks up right after the last event read, even from another process + for await (const event of hatchet.streams.events(topic, { cursor })) { + console.log(decode(event.payload)); + } +} + +export async function stopWithSignal(topic: string) { + // > Stop reading + const controller = new AbortController(); + setTimeout(() => controller.abort(), 10_000); + + for await (const event of hatchet.streams.events(topic, { signal: controller.signal })) { + console.log(decode(event.payload)); + } +} + +export async function namespaces() { + // > Namespaces + await hatchet.streams.publish('updates', 'hello', { namespace: 'customer-a' }); + + for await (const event of hatchet.streams.events('updates', { namespace: 'customer-a' })) { + console.log(decode(event.payload)); + break; + } +} + +export async function resetTopic() { + // > Reset a topic + await hatchet.streams.reset('orders'); + + // starts after the reset; readers that were already iterating aren't affected + for await (const event of hatchet.streams.events('orders')) { + console.log(decode(event.payload)); + } +} + +async function main() { + await consume(); + await publishFromAnywhere(); + await stopWithSignal('orders'); + await namespaces(); +} + +if (require.main === module) { + main() + .catch(console.error) + .finally(() => { + process.exit(0); + }); +} diff --git a/examples/typescript/durable-streams/worker.ts b/examples/typescript/durable-streams/worker.ts new file mode 100644 index 0000000000..6379db23d8 --- /dev/null +++ b/examples/typescript/durable-streams/worker.ts @@ -0,0 +1,14 @@ +import { hatchet } from '../hatchet-client'; +import { progressTask } from './workflow'; + +async function main() { + const worker = await hatchet.worker('durable-streams-worker', { + workflows: [progressTask], + }); + + await worker.start(); +} + +if (require.main === module) { + main(); +} diff --git a/examples/typescript/durable-streams/workflow.ts b/examples/typescript/durable-streams/workflow.ts new file mode 100644 index 0000000000..965a7f001f --- /dev/null +++ b/examples/typescript/durable-streams/workflow.ts @@ -0,0 +1,21 @@ +import sleep from '../../../util/sleep'; +import { hatchet } from '../hatchet-client'; + +// > Publishing +export type ProgressInput = { + topic: string; +}; + +export const progressTask = hatchet.task({ + name: 'durable-streams-progress', + fn: async (input: ProgressInput) => { + const total = 5; + + for (let step = 1; step <= total; step += 1) { + await sleep(500); + await hatchet.streams.publish(input.topic, JSON.stringify({ step, total })); + } + + return { steps: total }; + }, +}); diff --git a/frontend/docs/content/docs/reference/changelog/typescript.mdx b/frontend/docs/content/docs/reference/changelog/typescript.mdx index d78e56a56c..ae6dfdca7a 100644 --- a/frontend/docs/content/docs/reference/changelog/typescript.mdx +++ b/frontend/docs/content/docs/reference/changelog/typescript.mdx @@ -6,6 +6,12 @@ description: "Release notes for every Hatchet TypeScript SDK release, newest fir {/* AUTOGENERATED — do not edit. Run `task sync-changelog` to regenerate from sdks/typescript/CHANGELOG.md */} +## v1.38.0 - 2026-10-13 + +### Added + +- Added `hatchet.streams.reset(topic, { namespace })`, after which `events()` calls without a cursor start after the topic's latest message. Readers already iterating, and calls with a cursor, are unaffected. + ## v1.37.0 - 2026-10-07 ### Added diff --git a/frontend/docs/content/docs/reference/typescript/feature-clients/streams.mdx b/frontend/docs/content/docs/reference/typescript/feature-clients/streams.mdx index 96af4e88a5..24f4f971a0 100644 --- a/frontend/docs/content/docs/reference/typescript/feature-clients/streams.mdx +++ b/frontend/docs/content/docs/reference/typescript/feature-clients/streams.mdx @@ -16,7 +16,8 @@ readers can resume from a cursor. ##### `events()` -Yields topic's messages after options.cursor, or from the oldest retained. +Yields topic's messages after options.cursor, or from the oldest retained +(or, once the topic is reset, after the reset). If the connection drops, it resubscribes from the last message yielded, so nothing is skipped or repeated. Ends when the server hangs up an idle subscription, options.signal aborts, or the caller stops iterating; throws @@ -52,6 +53,25 @@ Returns `Promise`\<`void`\> + + +##### `reset()` + +Makes events() calls without a cursor start after the topic's latest +message. Readers already iterating, and any call with a cursor, are +unaffected. + +Parameters + +| Parameter | Type | Description | +| ---------- | -------------------------------------------- | --------------------------- | +| `topic` | `string` | the topic to reset | +| `options?` | `Pick`\<`StreamCallOptions`, `"namespace"`\> | optional namespace override | + +Returns + +`Promise`\<`void`\> + ##### `topicMetadata()` diff --git a/internal/services/shared/proto/v1/streams.pb.go b/internal/services/shared/proto/v1/streams.pb.go index 324af29063..795fdafb4b 100644 --- a/internal/services/shared/proto/v1/streams.pb.go +++ b/internal/services/shared/proto/v1/streams.pb.go @@ -491,6 +491,99 @@ func (x *StreamTopicMetadata) GetLastPublishedAt() *timestamppb.Timestamp { return nil } +type ResetStreamTopicRequest struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Namespace string `protobuf:"bytes,1,opt,name=namespace,proto3" json:"namespace,omitempty"` + Topic string `protobuf:"bytes,2,opt,name=topic,proto3" json:"topic,omitempty"` +} + +func (x *ResetStreamTopicRequest) Reset() { + *x = ResetStreamTopicRequest{} + if protoimpl.UnsafeEnabled { + mi := &file_v1_streams_proto_msgTypes[7] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *ResetStreamTopicRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ResetStreamTopicRequest) ProtoMessage() {} + +func (x *ResetStreamTopicRequest) ProtoReflect() protoreflect.Message { + mi := &file_v1_streams_proto_msgTypes[7] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ResetStreamTopicRequest.ProtoReflect.Descriptor instead. +func (*ResetStreamTopicRequest) Descriptor() ([]byte, []int) { + return file_v1_streams_proto_rawDescGZIP(), []int{7} +} + +func (x *ResetStreamTopicRequest) GetNamespace() string { + if x != nil { + return x.Namespace + } + return "" +} + +func (x *ResetStreamTopicRequest) GetTopic() string { + if x != nil { + return x.Topic + } + return "" +} + +type ResetStreamTopicResponse struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields +} + +func (x *ResetStreamTopicResponse) Reset() { + *x = ResetStreamTopicResponse{} + if protoimpl.UnsafeEnabled { + mi := &file_v1_streams_proto_msgTypes[8] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *ResetStreamTopicResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ResetStreamTopicResponse) ProtoMessage() {} + +func (x *ResetStreamTopicResponse) ProtoReflect() protoreflect.Message { + mi := &file_v1_streams_proto_msgTypes[8] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ResetStreamTopicResponse.ProtoReflect.Descriptor instead. +func (*ResetStreamTopicResponse) Descriptor() ([]byte, []int) { + return file_v1_streams_proto_rawDescGZIP(), []int{8} +} + var File_v1_streams_proto protoreflect.FileDescriptor var file_v1_streams_proto_rawDesc = []byte{ @@ -559,26 +652,37 @@ var file_v1_streams_proto_rawDesc = []byte{ 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x54, 0x69, 0x6d, 0x65, 0x73, 0x74, 0x61, 0x6d, 0x70, 0x52, 0x0f, 0x6c, 0x61, 0x73, 0x74, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x65, 0x64, 0x41, 0x74, 0x42, 0x10, 0x0a, 0x0e, 0x5f, 0x6c, 0x61, 0x74, 0x65, 0x73, 0x74, 0x5f, 0x63, 0x75, 0x72, 0x73, 0x6f, - 0x72, 0x32, 0xed, 0x01, 0x0a, 0x09, 0x56, 0x31, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x73, 0x12, - 0x4e, 0x0a, 0x07, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x12, 0x1f, 0x2e, 0x76, 0x31, 0x2e, - 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4d, 0x65, 0x73, - 0x73, 0x61, 0x67, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x20, 0x2e, 0x76, 0x31, - 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, 0x73, 0x68, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4d, 0x65, - 0x73, 0x73, 0x61, 0x67, 0x65, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, - 0x3e, 0x0a, 0x09, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x12, 0x1a, 0x2e, 0x76, - 0x31, 0x2e, 0x53, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x53, 0x74, 0x72, 0x65, 0x61, - 0x6d, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x11, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x74, - 0x72, 0x65, 0x61, 0x6d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x00, 0x30, 0x01, 0x12, - 0x50, 0x0a, 0x10, 0x47, 0x65, 0x74, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x4d, 0x65, 0x74, 0x61, 0x64, - 0x61, 0x74, 0x61, 0x12, 0x21, 0x2e, 0x76, 0x31, 0x2e, 0x47, 0x65, 0x74, 0x53, 0x74, 0x72, 0x65, - 0x61, 0x6d, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x52, - 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x17, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x74, 0x72, 0x65, - 0x61, 0x6d, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x22, - 0x00, 0x42, 0x42, 0x5a, 0x40, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, - 0x68, 0x61, 0x74, 0x63, 0x68, 0x65, 0x74, 0x2d, 0x64, 0x65, 0x76, 0x2f, 0x68, 0x61, 0x74, 0x63, - 0x68, 0x65, 0x74, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x65, 0x72, - 0x76, 0x69, 0x63, 0x65, 0x73, 0x2f, 0x73, 0x68, 0x61, 0x72, 0x65, 0x64, 0x2f, 0x70, 0x72, 0x6f, - 0x74, 0x6f, 0x2f, 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x72, 0x22, 0x4d, 0x0a, 0x17, 0x52, 0x65, 0x73, 0x65, 0x74, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, + 0x54, 0x6f, 0x70, 0x69, 0x63, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1c, 0x0a, 0x09, + 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, + 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x12, 0x14, 0x0a, 0x05, 0x74, 0x6f, + 0x70, 0x69, 0x63, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x74, 0x6f, 0x70, 0x69, 0x63, + 0x22, 0x1a, 0x0a, 0x18, 0x52, 0x65, 0x73, 0x65, 0x74, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x54, + 0x6f, 0x70, 0x69, 0x63, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x32, 0xb3, 0x02, 0x0a, + 0x09, 0x56, 0x31, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x73, 0x12, 0x4e, 0x0a, 0x07, 0x50, 0x75, + 0x62, 0x6c, 0x69, 0x73, 0x68, 0x12, 0x1f, 0x2e, 0x76, 0x31, 0x2e, 0x50, 0x75, 0x62, 0x6c, 0x69, + 0x73, 0x68, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x52, + 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x20, 0x2e, 0x76, 0x31, 0x2e, 0x50, 0x75, 0x62, 0x6c, + 0x69, 0x73, 0x68, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, + 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x3e, 0x0a, 0x09, 0x53, 0x75, + 0x62, 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x12, 0x1a, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x75, 0x62, + 0x73, 0x63, 0x72, 0x69, 0x62, 0x65, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x52, 0x65, 0x71, 0x75, + 0x65, 0x73, 0x74, 0x1a, 0x11, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x4d, + 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x22, 0x00, 0x30, 0x01, 0x12, 0x50, 0x0a, 0x10, 0x47, 0x65, + 0x74, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x12, 0x21, + 0x2e, 0x76, 0x31, 0x2e, 0x47, 0x65, 0x74, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x54, 0x6f, 0x70, + 0x69, 0x63, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, + 0x74, 0x1a, 0x17, 0x2e, 0x76, 0x31, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x54, 0x6f, 0x70, + 0x69, 0x63, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x22, 0x00, 0x12, 0x44, 0x0a, 0x05, + 0x52, 0x65, 0x73, 0x65, 0x74, 0x12, 0x1b, 0x2e, 0x76, 0x31, 0x2e, 0x52, 0x65, 0x73, 0x65, 0x74, + 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x52, 0x65, 0x71, 0x75, 0x65, + 0x73, 0x74, 0x1a, 0x1c, 0x2e, 0x76, 0x31, 0x2e, 0x52, 0x65, 0x73, 0x65, 0x74, 0x53, 0x74, 0x72, + 0x65, 0x61, 0x6d, 0x54, 0x6f, 0x70, 0x69, 0x63, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, + 0x22, 0x00, 0x42, 0x42, 0x5a, 0x40, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, + 0x2f, 0x68, 0x61, 0x74, 0x63, 0x68, 0x65, 0x74, 0x2d, 0x64, 0x65, 0x76, 0x2f, 0x68, 0x61, 0x74, + 0x63, 0x68, 0x65, 0x74, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x73, 0x65, + 0x72, 0x76, 0x69, 0x63, 0x65, 0x73, 0x2f, 0x73, 0x68, 0x61, 0x72, 0x65, 0x64, 0x2f, 0x70, 0x72, + 0x6f, 0x74, 0x6f, 0x2f, 0x76, 0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( @@ -593,7 +697,7 @@ func file_v1_streams_proto_rawDescGZIP() []byte { return file_v1_streams_proto_rawDescData } -var file_v1_streams_proto_msgTypes = make([]protoimpl.MessageInfo, 7) +var file_v1_streams_proto_msgTypes = make([]protoimpl.MessageInfo, 9) var file_v1_streams_proto_goTypes = []interface{}{ (*PublishStreamMessageRequest)(nil), // 0: v1.PublishStreamMessageRequest (*PublishStreamMessageResponse)(nil), // 1: v1.PublishStreamMessageResponse @@ -602,20 +706,24 @@ var file_v1_streams_proto_goTypes = []interface{}{ (*StreamEntry)(nil), // 4: v1.StreamEntry (*GetStreamTopicMetadataRequest)(nil), // 5: v1.GetStreamTopicMetadataRequest (*StreamTopicMetadata)(nil), // 6: v1.StreamTopicMetadata - (*timestamppb.Timestamp)(nil), // 7: google.protobuf.Timestamp + (*ResetStreamTopicRequest)(nil), // 7: v1.ResetStreamTopicRequest + (*ResetStreamTopicResponse)(nil), // 8: v1.ResetStreamTopicResponse + (*timestamppb.Timestamp)(nil), // 9: google.protobuf.Timestamp } var file_v1_streams_proto_depIdxs = []int32{ 4, // 0: v1.StreamMessage.entries:type_name -> v1.StreamEntry - 7, // 1: v1.StreamEntry.created_at:type_name -> google.protobuf.Timestamp - 7, // 2: v1.StreamTopicMetadata.last_published_at:type_name -> google.protobuf.Timestamp + 9, // 1: v1.StreamEntry.created_at:type_name -> google.protobuf.Timestamp + 9, // 2: v1.StreamTopicMetadata.last_published_at:type_name -> google.protobuf.Timestamp 0, // 3: v1.V1Streams.Publish:input_type -> v1.PublishStreamMessageRequest 2, // 4: v1.V1Streams.Subscribe:input_type -> v1.SubscribeStreamRequest 5, // 5: v1.V1Streams.GetTopicMetadata:input_type -> v1.GetStreamTopicMetadataRequest - 1, // 6: v1.V1Streams.Publish:output_type -> v1.PublishStreamMessageResponse - 3, // 7: v1.V1Streams.Subscribe:output_type -> v1.StreamMessage - 6, // 8: v1.V1Streams.GetTopicMetadata:output_type -> v1.StreamTopicMetadata - 6, // [6:9] is the sub-list for method output_type - 3, // [3:6] is the sub-list for method input_type + 7, // 6: v1.V1Streams.Reset:input_type -> v1.ResetStreamTopicRequest + 1, // 7: v1.V1Streams.Publish:output_type -> v1.PublishStreamMessageResponse + 3, // 8: v1.V1Streams.Subscribe:output_type -> v1.StreamMessage + 6, // 9: v1.V1Streams.GetTopicMetadata:output_type -> v1.StreamTopicMetadata + 8, // 10: v1.V1Streams.Reset:output_type -> v1.ResetStreamTopicResponse + 7, // [7:11] is the sub-list for method output_type + 3, // [3:7] is the sub-list for method input_type 3, // [3:3] is the sub-list for extension type_name 3, // [3:3] is the sub-list for extension extendee 0, // [0:3] is the sub-list for field type_name @@ -711,6 +819,30 @@ func file_v1_streams_proto_init() { return nil } } + file_v1_streams_proto_msgTypes[7].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*ResetStreamTopicRequest); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_v1_streams_proto_msgTypes[8].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*ResetStreamTopicResponse); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } } file_v1_streams_proto_msgTypes[2].OneofWrappers = []interface{}{} file_v1_streams_proto_msgTypes[6].OneofWrappers = []interface{}{} @@ -720,7 +852,7 @@ func file_v1_streams_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_v1_streams_proto_rawDesc, NumEnums: 0, - NumMessages: 7, + NumMessages: 9, NumExtensions: 0, NumServices: 1, }, diff --git a/internal/services/shared/proto/v1/streams_grpc.pb.go b/internal/services/shared/proto/v1/streams_grpc.pb.go index 2268ab1159..ff79b9ff73 100644 --- a/internal/services/shared/proto/v1/streams_grpc.pb.go +++ b/internal/services/shared/proto/v1/streams_grpc.pb.go @@ -25,6 +25,7 @@ type V1StreamsClient interface { Publish(ctx context.Context, in *PublishStreamMessageRequest, opts ...grpc.CallOption) (*PublishStreamMessageResponse, error) Subscribe(ctx context.Context, in *SubscribeStreamRequest, opts ...grpc.CallOption) (V1Streams_SubscribeClient, error) GetTopicMetadata(ctx context.Context, in *GetStreamTopicMetadataRequest, opts ...grpc.CallOption) (*StreamTopicMetadata, error) + Reset(ctx context.Context, in *ResetStreamTopicRequest, opts ...grpc.CallOption) (*ResetStreamTopicResponse, error) } type v1StreamsClient struct { @@ -85,6 +86,15 @@ func (c *v1StreamsClient) GetTopicMetadata(ctx context.Context, in *GetStreamTop return out, nil } +func (c *v1StreamsClient) Reset(ctx context.Context, in *ResetStreamTopicRequest, opts ...grpc.CallOption) (*ResetStreamTopicResponse, error) { + out := new(ResetStreamTopicResponse) + err := c.cc.Invoke(ctx, "/v1.V1Streams/Reset", in, out, opts...) + if err != nil { + return nil, err + } + return out, nil +} + // V1StreamsServer is the server API for V1Streams service. // All implementations must embed UnimplementedV1StreamsServer // for forward compatibility @@ -92,6 +102,7 @@ type V1StreamsServer interface { Publish(context.Context, *PublishStreamMessageRequest) (*PublishStreamMessageResponse, error) Subscribe(*SubscribeStreamRequest, V1Streams_SubscribeServer) error GetTopicMetadata(context.Context, *GetStreamTopicMetadataRequest) (*StreamTopicMetadata, error) + Reset(context.Context, *ResetStreamTopicRequest) (*ResetStreamTopicResponse, error) mustEmbedUnimplementedV1StreamsServer() } @@ -108,6 +119,9 @@ func (UnimplementedV1StreamsServer) Subscribe(*SubscribeStreamRequest, V1Streams func (UnimplementedV1StreamsServer) GetTopicMetadata(context.Context, *GetStreamTopicMetadataRequest) (*StreamTopicMetadata, error) { return nil, status.Errorf(codes.Unimplemented, "method GetTopicMetadata not implemented") } +func (UnimplementedV1StreamsServer) Reset(context.Context, *ResetStreamTopicRequest) (*ResetStreamTopicResponse, error) { + return nil, status.Errorf(codes.Unimplemented, "method Reset not implemented") +} func (UnimplementedV1StreamsServer) mustEmbedUnimplementedV1StreamsServer() {} // UnsafeV1StreamsServer may be embedded to opt out of forward compatibility for this service. @@ -178,6 +192,24 @@ func _V1Streams_GetTopicMetadata_Handler(srv interface{}, ctx context.Context, d return interceptor(ctx, in, info, handler) } +func _V1Streams_Reset_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(ResetStreamTopicRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(V1StreamsServer).Reset(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: "/v1.V1Streams/Reset", + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(V1StreamsServer).Reset(ctx, req.(*ResetStreamTopicRequest)) + } + return interceptor(ctx, in, info, handler) +} + // V1Streams_ServiceDesc is the grpc.ServiceDesc for V1Streams service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -193,6 +225,10 @@ var V1Streams_ServiceDesc = grpc.ServiceDesc{ MethodName: "GetTopicMetadata", Handler: _V1Streams_GetTopicMetadata_Handler, }, + { + MethodName: "Reset", + Handler: _V1Streams_Reset_Handler, + }, }, Streams: []grpc.StreamDesc{ { diff --git a/internal/services/shared/proto/v1/v1connect/streams.connect.go b/internal/services/shared/proto/v1/v1connect/streams.connect.go index cc1e630f19..7b3027c84b 100644 --- a/internal/services/shared/proto/v1/v1connect/streams.connect.go +++ b/internal/services/shared/proto/v1/v1connect/streams.connect.go @@ -40,6 +40,8 @@ const ( // V1StreamsGetTopicMetadataProcedure is the fully-qualified name of the V1Streams's // GetTopicMetadata RPC. V1StreamsGetTopicMetadataProcedure = "/v1.V1Streams/GetTopicMetadata" + // V1StreamsResetProcedure is the fully-qualified name of the V1Streams's Reset RPC. + V1StreamsResetProcedure = "/v1.V1Streams/Reset" ) // V1StreamsClient is a client for the v1.V1Streams service. @@ -47,6 +49,7 @@ type V1StreamsClient interface { Publish(context.Context, *v1.PublishStreamMessageRequest) (*v1.PublishStreamMessageResponse, error) Subscribe(context.Context, *v1.SubscribeStreamRequest) (*connect.ServerStreamForClient[v1.StreamMessage], error) GetTopicMetadata(context.Context, *v1.GetStreamTopicMetadataRequest) (*v1.StreamTopicMetadata, error) + Reset(context.Context, *v1.ResetStreamTopicRequest) (*v1.ResetStreamTopicResponse, error) } // NewV1StreamsClient constructs a client for the v1.V1Streams service. By default, it uses the @@ -78,6 +81,12 @@ func NewV1StreamsClient(httpClient connect.HTTPClient, baseURL string, opts ...c connect.WithSchema(v1StreamsMethods.ByName("GetTopicMetadata")), connect.WithClientOptions(opts...), ), + reset: connect.NewClient[v1.ResetStreamTopicRequest, v1.ResetStreamTopicResponse]( + httpClient, + baseURL+V1StreamsResetProcedure, + connect.WithSchema(v1StreamsMethods.ByName("Reset")), + connect.WithClientOptions(opts...), + ), } } @@ -86,6 +95,7 @@ type v1StreamsClient struct { publish *connect.Client[v1.PublishStreamMessageRequest, v1.PublishStreamMessageResponse] subscribe *connect.Client[v1.SubscribeStreamRequest, v1.StreamMessage] getTopicMetadata *connect.Client[v1.GetStreamTopicMetadataRequest, v1.StreamTopicMetadata] + reset *connect.Client[v1.ResetStreamTopicRequest, v1.ResetStreamTopicResponse] } // Publish calls v1.V1Streams.Publish. @@ -111,11 +121,21 @@ func (c *v1StreamsClient) GetTopicMetadata(ctx context.Context, req *v1.GetStrea return nil, err } +// Reset calls v1.V1Streams.Reset. +func (c *v1StreamsClient) Reset(ctx context.Context, req *v1.ResetStreamTopicRequest) (*v1.ResetStreamTopicResponse, error) { + response, err := c.reset.CallUnary(ctx, connect.NewRequest(req)) + if response != nil { + return response.Msg, err + } + return nil, err +} + // V1StreamsHandler is an implementation of the v1.V1Streams service. type V1StreamsHandler interface { Publish(context.Context, *v1.PublishStreamMessageRequest) (*v1.PublishStreamMessageResponse, error) Subscribe(context.Context, *v1.SubscribeStreamRequest, *connect.ServerStream[v1.StreamMessage]) error GetTopicMetadata(context.Context, *v1.GetStreamTopicMetadataRequest) (*v1.StreamTopicMetadata, error) + Reset(context.Context, *v1.ResetStreamTopicRequest) (*v1.ResetStreamTopicResponse, error) } // NewV1StreamsHandler builds an HTTP handler from the service implementation. It returns the path @@ -143,6 +163,12 @@ func NewV1StreamsHandler(svc V1StreamsHandler, opts ...connect.HandlerOption) (s connect.WithSchema(v1StreamsMethods.ByName("GetTopicMetadata")), connect.WithHandlerOptions(opts...), ) + v1StreamsResetHandler := connect.NewUnaryHandlerSimple( + V1StreamsResetProcedure, + svc.Reset, + connect.WithSchema(v1StreamsMethods.ByName("Reset")), + connect.WithHandlerOptions(opts...), + ) return "/v1.V1Streams/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { case V1StreamsPublishProcedure: @@ -151,6 +177,8 @@ func NewV1StreamsHandler(svc V1StreamsHandler, opts ...connect.HandlerOption) (s v1StreamsSubscribeHandler.ServeHTTP(w, r) case V1StreamsGetTopicMetadataProcedure: v1StreamsGetTopicMetadataHandler.ServeHTTP(w, r) + case V1StreamsResetProcedure: + v1StreamsResetHandler.ServeHTTP(w, r) default: http.NotFound(w, r) } @@ -171,3 +199,7 @@ func (UnimplementedV1StreamsHandler) Subscribe(context.Context, *v1.SubscribeStr func (UnimplementedV1StreamsHandler) GetTopicMetadata(context.Context, *v1.GetStreamTopicMetadataRequest) (*v1.StreamTopicMetadata, error) { return nil, connect.NewError(connect.CodeUnimplemented, errors.New("v1.V1Streams.GetTopicMetadata is not implemented")) } + +func (UnimplementedV1StreamsHandler) Reset(context.Context, *v1.ResetStreamTopicRequest) (*v1.ResetStreamTopicResponse, error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("v1.V1Streams.Reset is not implemented")) +} diff --git a/internal/services/streams/server.go b/internal/services/streams/server.go index ff93889ed8..f9d1735f5a 100644 --- a/internal/services/streams/server.go +++ b/internal/services/streams/server.go @@ -175,6 +175,23 @@ func (s *ServiceImpl) Subscribe(ctx context.Context, req *contracts.SubscribeStr telemetry.AttributeKV{Key: "stream.resumed", Value: cursor.ID > 0}, ) + if req.Cursor == nil || *req.Cursor == "" { + if cursor, err = s.repo.Streams().TopicStartCursor(ctx, tenantId, namespace, topic); err != nil { + return err + } + } + + encodedStart, err := v1.EncodeStreamCursor(cursor) + + if err != nil { + return err + } + + // a client that reconnects before any entry arrives resumes here, not from a later reset + if err := sender.Send(&contracts.StreamMessage{Cursor: encodedStart}); err != nil { + return err + } + ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -210,6 +227,37 @@ func (s *ServiceImpl) Subscribe(ctx context.Context, req *contracts.SubscribeStr return nil } +func (s *ServiceImpl) Reset(ctx context.Context, req *contracts.ResetStreamTopicRequest) (*contracts.ResetStreamTopicResponse, error) { + tenant := ctx.Value("tenant").(*sqlcv1.Tenant) + tenantId := tenant.ID + + telemetry.WithAttributes(trace.SpanFromContext(ctx), + telemetry.AttributeKV{Key: "tenant.id", Value: tenantId}, + telemetry.AttributeKV{Key: "stream.namespace", Value: req.Namespace}, + telemetry.AttributeKV{Key: "stream.topic", Value: req.Topic}, + ) + + entitled, err := s.repo.TenantEntitlement().HasEntitlement(ctx, tenantId, v1.EntitlementDurableStreams) + + if err != nil { + return nil, err + } + + if !entitled { + return nil, connect.NewError(connect.CodePermissionDenied, errNotEntitled) + } + + if err := v1.ValidateStreamAddress(req.Namespace, req.Topic); err != nil { + return nil, connect.NewError(connect.CodeInvalidArgument, err) + } + + if err := s.repo.Streams().ResetTopic(ctx, tenantId, req.Namespace, req.Topic); err != nil { + return nil, err + } + + return &contracts.ResetStreamTopicResponse{}, nil +} + func (s *ServiceImpl) GetTopicMetadata(ctx context.Context, req *contracts.GetStreamTopicMetadataRequest) (*contracts.StreamTopicMetadata, error) { tenant := ctx.Value("tenant").(*sqlcv1.Tenant) tenantId := tenant.ID diff --git a/pkg/repository/sqlcv1/models.go b/pkg/repository/sqlcv1/models.go index d716000aaa..3cd3fc62f5 100644 --- a/pkg/repository/sqlcv1/models.go +++ b/pkg/repository/sqlcv1/models.go @@ -3888,6 +3888,7 @@ type V1StreamTopic struct { InsertedAt pgtype.Timestamptz `json:"inserted_at"` LastPublishedAt pgtype.Timestamptz `json:"last_published_at"` LastOffset int64 `json:"last_offset"` + ResetOffset int64 `json:"reset_offset"` } type V1Task struct { diff --git a/pkg/repository/sqlcv1/streams.sql b/pkg/repository/sqlcv1/streams.sql index 5dc567b0fc..81e61785f5 100644 --- a/pkg/repository/sqlcv1/streams.sql +++ b/pkg/repository/sqlcv1/streams.sql @@ -273,3 +273,15 @@ SELECT FROM retained LEFT JOIN latest ON true WHERE EXISTS (SELECT 1 FROM topic); + +-- name: ResetStreamTopic :exec +-- Waits on the row lock a publish holds until commit (see +-- ReserveStreamTopicOffsets), so every offset up to last_offset is stored. +UPDATE v1_stream_topic +SET reset_offset = last_offset +WHERE tenant_id = @tenantId::uuid AND namespace = @namespace::text AND topic = @topic::text; + +-- name: GetStreamTopicResetOffset :one +SELECT reset_offset +FROM v1_stream_topic +WHERE tenant_id = @tenantId::uuid AND namespace = @namespace::text AND topic = @topic::text; diff --git a/pkg/repository/sqlcv1/streams.sql.go b/pkg/repository/sqlcv1/streams.sql.go index 2560f6977b..05c16079cf 100644 --- a/pkg/repository/sqlcv1/streams.sql.go +++ b/pkg/repository/sqlcv1/streams.sql.go @@ -282,6 +282,25 @@ func (q *Queries) GetStreamTopicMetadata(ctx context.Context, db DBTX, arg GetSt return &i, err } +const getStreamTopicResetOffset = `-- name: GetStreamTopicResetOffset :one +SELECT reset_offset +FROM v1_stream_topic +WHERE tenant_id = $1::uuid AND namespace = $2::text AND topic = $3::text +` + +type GetStreamTopicResetOffsetParams struct { + Tenantid uuid.UUID `json:"tenantid"` + Namespace string `json:"namespace"` + Topic string `json:"topic"` +} + +func (q *Queries) GetStreamTopicResetOffset(ctx context.Context, db DBTX, arg GetStreamTopicResetOffsetParams) (int64, error) { + row := db.QueryRow(ctx, getStreamTopicResetOffset, arg.Tenantid, arg.Namespace, arg.Topic) + var reset_offset int64 + err := row.Scan(&reset_offset) + return reset_offset, err +} + const insertStreamPayload = `-- name: InsertStreamPayload :one INSERT INTO v1_stream_payload (id, tenant_id, payload) VALUES ($1::uuid, $2::uuid, $3::bytea) @@ -522,6 +541,25 @@ func (q *Queries) ListStreamRetentionDeleteCandidates(ctx context.Context, db DB return items, nil } +const resetStreamTopic = `-- name: ResetStreamTopic :exec +UPDATE v1_stream_topic +SET reset_offset = last_offset +WHERE tenant_id = $1::uuid AND namespace = $2::text AND topic = $3::text +` + +type ResetStreamTopicParams struct { + Tenantid uuid.UUID `json:"tenantid"` + Namespace string `json:"namespace"` + Topic string `json:"topic"` +} + +// Waits on the row lock a publish holds until commit (see +// ReserveStreamTopicOffsets), so every offset up to last_offset is stored. +func (q *Queries) ResetStreamTopic(ctx context.Context, db DBTX, arg ResetStreamTopicParams) error { + _, err := db.Exec(ctx, resetStreamTopic, arg.Tenantid, arg.Namespace, arg.Topic) + return err +} + const setStreamMessagePayloadRef = `-- name: SetStreamMessagePayloadRef :exec UPDATE v1_stream_message SET payload_id = $1::uuid, payload_inserted_at = $2::timestamptz @@ -581,7 +619,7 @@ INSERT INTO v1_stream_topic (tenant_id, namespace, topic) VALUES ($1::uuid, $2::text, $3::text) ON CONFLICT (tenant_id, namespace, topic) DO UPDATE SET last_published_at = NOW() -RETURNING v1_stream_topic.tenant_id, v1_stream_topic.namespace, v1_stream_topic.topic, v1_stream_topic.inserted_at, v1_stream_topic.last_published_at, v1_stream_topic.last_offset, (xmax = 0) AS inserted +RETURNING v1_stream_topic.tenant_id, v1_stream_topic.namespace, v1_stream_topic.topic, v1_stream_topic.inserted_at, v1_stream_topic.last_published_at, v1_stream_topic.last_offset, v1_stream_topic.reset_offset, (xmax = 0) AS inserted ` type UpsertStreamTopicParams struct { @@ -606,6 +644,7 @@ func (q *Queries) UpsertStreamTopic(ctx context.Context, db DBTX, arg UpsertStre &i.V1StreamTopic.InsertedAt, &i.V1StreamTopic.LastPublishedAt, &i.V1StreamTopic.LastOffset, + &i.V1StreamTopic.ResetOffset, &i.Inserted, ) return &i, err diff --git a/pkg/repository/streams.go b/pkg/repository/streams.go index c0211470fd..b1fb296837 100644 --- a/pkg/repository/streams.go +++ b/pkg/repository/streams.go @@ -175,6 +175,14 @@ type StreamsRepository interface { // CheckCursorRetained returns a *StreamCursorExpiredError if cursor is // older than the tenant's retention or the oldest partition. CheckCursorRetained(ctx context.Context, tenantId uuid.UUID, cursor StreamCursor) error + + // ResetTopic moves where subscribers without a cursor start to after the + // topic's latest stored message. A topic never published to is left as is. + ResetTopic(ctx context.Context, tenantId uuid.UUID, namespace, topic string) error + + // TopicStartCursor is where a subscriber without a cursor starts: after the + // last reset, else the topic's beginning. + TopicStartCursor(ctx context.Context, tenantId uuid.UUID, namespace, topic string) (StreamCursor, error) } // StreamCursorExpiredError: resuming from the cursor would skip deleted messages. @@ -560,8 +568,9 @@ func (r *streamsRepositoryImpl) GetTopicMetadata(ctx context.Context, tenantId u } func (r *streamsRepositoryImpl) CheckCursorRetained(ctx context.Context, tenantId uuid.UUID, cursor StreamCursor) error { - // the zero cursor means "from the oldest retained message", which can't expire - if cursor.ID == 0 { + // cursors without a time come from the topic's start or a reset, which can't + // expire: they mean "from the oldest retained message after this offset" + if cursor.CreatedAt.IsZero() { return nil } @@ -589,3 +598,26 @@ func (r *streamsRepositoryImpl) CheckCursorRetained(ctx context.Context, tenantI return nil } + +func (r *streamsRepositoryImpl) ResetTopic(ctx context.Context, tenantId uuid.UUID, namespace, topic string) error { + return r.queries.ResetStreamTopic(ctx, r.pool, sqlcv1.ResetStreamTopicParams{ + Tenantid: tenantId, + Namespace: namespace, + Topic: topic, + }) +} + +func (r *streamsRepositoryImpl) TopicStartCursor(ctx context.Context, tenantId uuid.UUID, namespace, topic string) (StreamCursor, error) { + resetOffset, err := r.queries.GetStreamTopicResetOffset(ctx, r.pool, sqlcv1.GetStreamTopicResetOffsetParams{ + Tenantid: tenantId, + Namespace: namespace, + Topic: topic, + }) + + if err != nil && !errors.Is(err, pgx.ErrNoRows) { + return StreamCursor{}, err + } + + // offsets start at 1, so ID 0 is the start of the topic + return StreamCursor{Namespace: namespace, Topic: topic, ID: resetOffset}, nil +} diff --git a/pkg/repository/streams_repository_test.go b/pkg/repository/streams_repository_test.go index 35259ef54c..f933263dd5 100644 --- a/pkg/repository/streams_repository_test.go +++ b/pkg/repository/streams_repository_test.go @@ -450,4 +450,37 @@ func TestStreamsRepository(t *testing.T) { require.NoError(t, err) assert.False(t, dup[0].Inserted) }) + + t.Run("reset moves where cursorless subscribers start, not earlier cursors", func(t *testing.T) { + repo := createStreamsRepository(t, pool) + tenantId := uuid.New() + + require.NoError(t, repo.ResetTopic(ctx, tenantId, "", "t"), "resetting a topic never published to is a no-op") + + for seq, payload := range []string{"one", "two"} { + _, err := repo.InsertOrderedStreamMessage(ctx, tenantId, streamOpts(int64(seq), payload)) + require.NoError(t, err) + } + + before, err := repo.TopicStartCursor(ctx, tenantId, "", "t") + require.NoError(t, err) + assert.Equal(t, int64(0), before.ID, "a topic never reset starts at its beginning") + + require.NoError(t, repo.ResetTopic(ctx, tenantId, "", "t")) + + _, err = repo.InsertOrderedStreamMessage(ctx, tenantId, streamOpts(2, "three")) + require.NoError(t, err) + + after, err := repo.TopicStartCursor(ctx, tenantId, "", "t") + require.NoError(t, err) + require.NoError(t, repo.CheckCursorRetained(ctx, tenantId, after), "a reset cursor can't expire") + + fromReset, err := repo.ListMessagesAfterCursor(ctx, tenantId, ListStreamMessagesOpts{Topic: "t", Cursor: after}) + require.NoError(t, err) + assert.Equal(t, []string{"three"}, streamPayloads(t, fromReset)) + + fromBefore, err := repo.ListMessagesAfterCursor(ctx, tenantId, ListStreamMessagesOpts{Topic: "t", Cursor: before}) + require.NoError(t, err) + assert.Equal(t, []string{"one", "two", "three"}, streamPayloads(t, fromBefore)) + }) } diff --git a/sdks/typescript/CHANGELOG.md b/sdks/typescript/CHANGELOG.md index a48dc9e93b..0a7fde1b22 100644 --- a/sdks/typescript/CHANGELOG.md +++ b/sdks/typescript/CHANGELOG.md @@ -5,6 +5,12 @@ All notable changes to Hatchet's TypeScript SDK will be documented in this chang The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.38.0] - 2026-10-13 + +### Added + +- Added `hatchet.streams.reset(topic, { namespace })`, after which `events()` calls without a cursor start after the topic's latest message. Readers already iterating, and calls with a cursor, are unaffected. + ## [1.37.0] - 2026-10-07 ### Added diff --git a/sdks/typescript/package.json b/sdks/typescript/package.json index 8499166291..7d110034ee 100644 --- a/sdks/typescript/package.json +++ b/sdks/typescript/package.json @@ -1,6 +1,6 @@ { "name": "@hatchet-dev/typescript-sdk", - "version": "1.37.0", + "version": "1.38.0", "engines": { "node": ">=20" }, diff --git a/sdks/typescript/src/protoc-es/v1/streams_pb.ts b/sdks/typescript/src/protoc-es/v1/streams_pb.ts index 4d4f112202..0d12599144 100644 --- a/sdks/typescript/src/protoc-es/v1/streams_pb.ts +++ b/sdks/typescript/src/protoc-es/v1/streams_pb.ts @@ -12,7 +12,7 @@ import type { Message } from "@bufbuild/protobuf"; * Describes the file v1/streams.proto. */ export const file_v1_streams: GenFile = /*@__PURE__*/ - fileDesc("ChB2MS9zdHJlYW1zLnByb3RvEgJ2MSKQAQobUHVibGlzaFN0cmVhbU1lc3NhZ2VSZXF1ZXN0EhEKCW5hbWVzcGFjZRgBIAEoCRINCgV0b3BpYxgCIAEoCRIPCgdwYXlsb2FkGAMgASgMEhMKC3Byb2R1Y2VyX2lkGAQgASgJEhQKDHByb2R1Y2VyX3NlcRgFIAEoAxITCgtwYXlsb2FkX3JlZhgGIAEoCSIeChxQdWJsaXNoU3RyZWFtTWVzc2FnZVJlc3BvbnNlIloKFlN1YnNjcmliZVN0cmVhbVJlcXVlc3QSEQoJbmFtZXNwYWNlGAEgASgJEg0KBXRvcGljGAIgASgJEhMKBmN1cnNvchgDIAEoCUgAiAEBQgkKB19jdXJzb3IiUQoNU3RyZWFtTWVzc2FnZRIgCgdlbnRyaWVzGAEgAygLMg8udjEuU3RyZWFtRW50cnkSDgoGaGFuZ3VwGAIgASgIEg4KBmN1cnNvchgDIAEoCSJzCgtTdHJlYW1FbnRyeRIPCgdwYXlsb2FkGAEgASgMEg4KBmN1cnNvchgCIAEoCRIuCgpjcmVhdGVkX2F0GAMgASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBITCgtwYXlsb2FkX3JlZhgEIAEoCSJBCh1HZXRTdHJlYW1Ub3BpY01ldGFkYXRhUmVxdWVzdBIRCgluYW1lc3BhY2UYASABKAkSDQoFdG9waWMYAiABKAkixgEKE1N0cmVhbVRvcGljTWV0YWRhdGESEQoJbmFtZXNwYWNlGAEgASgJEg0KBXRvcGljGAIgASgJEhEKCXRlbmFudF9pZBgDIAEoCRIaCg1sYXRlc3RfY3Vyc29yGAQgASgJSACIAQESFQoNbWVzc2FnZV9jb3VudBgFIAEoAxI1ChFsYXN0X3B1Ymxpc2hlZF9hdBgGIAEoCzIaLmdvb2dsZS5wcm90b2J1Zi5UaW1lc3RhbXBCEAoOX2xhdGVzdF9jdXJzb3Iy7QEKCVYxU3RyZWFtcxJOCgdQdWJsaXNoEh8udjEuUHVibGlzaFN0cmVhbU1lc3NhZ2VSZXF1ZXN0GiAudjEuUHVibGlzaFN0cmVhbU1lc3NhZ2VSZXNwb25zZSIAEj4KCVN1YnNjcmliZRIaLnYxLlN1YnNjcmliZVN0cmVhbVJlcXVlc3QaES52MS5TdHJlYW1NZXNzYWdlIgAwARJQChBHZXRUb3BpY01ldGFkYXRhEiEudjEuR2V0U3RyZWFtVG9waWNNZXRhZGF0YVJlcXVlc3QaFy52MS5TdHJlYW1Ub3BpY01ldGFkYXRhIgBCQlpAZ2l0aHViLmNvbS9oYXRjaGV0LWRldi9oYXRjaGV0L2ludGVybmFsL3NlcnZpY2VzL3NoYXJlZC9wcm90by92MWIGcHJvdG8z", [file_google_protobuf_timestamp]); + fileDesc("ChB2MS9zdHJlYW1zLnByb3RvEgJ2MSKQAQobUHVibGlzaFN0cmVhbU1lc3NhZ2VSZXF1ZXN0EhEKCW5hbWVzcGFjZRgBIAEoCRINCgV0b3BpYxgCIAEoCRIPCgdwYXlsb2FkGAMgASgMEhMKC3Byb2R1Y2VyX2lkGAQgASgJEhQKDHByb2R1Y2VyX3NlcRgFIAEoAxITCgtwYXlsb2FkX3JlZhgGIAEoCSIeChxQdWJsaXNoU3RyZWFtTWVzc2FnZVJlc3BvbnNlIloKFlN1YnNjcmliZVN0cmVhbVJlcXVlc3QSEQoJbmFtZXNwYWNlGAEgASgJEg0KBXRvcGljGAIgASgJEhMKBmN1cnNvchgDIAEoCUgAiAEBQgkKB19jdXJzb3IiUQoNU3RyZWFtTWVzc2FnZRIgCgdlbnRyaWVzGAEgAygLMg8udjEuU3RyZWFtRW50cnkSDgoGaGFuZ3VwGAIgASgIEg4KBmN1cnNvchgDIAEoCSJzCgtTdHJlYW1FbnRyeRIPCgdwYXlsb2FkGAEgASgMEg4KBmN1cnNvchgCIAEoCRIuCgpjcmVhdGVkX2F0GAMgASgLMhouZ29vZ2xlLnByb3RvYnVmLlRpbWVzdGFtcBITCgtwYXlsb2FkX3JlZhgEIAEoCSJBCh1HZXRTdHJlYW1Ub3BpY01ldGFkYXRhUmVxdWVzdBIRCgluYW1lc3BhY2UYASABKAkSDQoFdG9waWMYAiABKAkixgEKE1N0cmVhbVRvcGljTWV0YWRhdGESEQoJbmFtZXNwYWNlGAEgASgJEg0KBXRvcGljGAIgASgJEhEKCXRlbmFudF9pZBgDIAEoCRIaCg1sYXRlc3RfY3Vyc29yGAQgASgJSACIAQESFQoNbWVzc2FnZV9jb3VudBgFIAEoAxI1ChFsYXN0X3B1Ymxpc2hlZF9hdBgGIAEoCzIaLmdvb2dsZS5wcm90b2J1Zi5UaW1lc3RhbXBCEAoOX2xhdGVzdF9jdXJzb3IiOwoXUmVzZXRTdHJlYW1Ub3BpY1JlcXVlc3QSEQoJbmFtZXNwYWNlGAEgASgJEg0KBXRvcGljGAIgASgJIhoKGFJlc2V0U3RyZWFtVG9waWNSZXNwb25zZTKzAgoJVjFTdHJlYW1zEk4KB1B1Ymxpc2gSHy52MS5QdWJsaXNoU3RyZWFtTWVzc2FnZVJlcXVlc3QaIC52MS5QdWJsaXNoU3RyZWFtTWVzc2FnZVJlc3BvbnNlIgASPgoJU3Vic2NyaWJlEhoudjEuU3Vic2NyaWJlU3RyZWFtUmVxdWVzdBoRLnYxLlN0cmVhbU1lc3NhZ2UiADABElAKEEdldFRvcGljTWV0YWRhdGESIS52MS5HZXRTdHJlYW1Ub3BpY01ldGFkYXRhUmVxdWVzdBoXLnYxLlN0cmVhbVRvcGljTWV0YWRhdGEiABJECgVSZXNldBIbLnYxLlJlc2V0U3RyZWFtVG9waWNSZXF1ZXN0GhwudjEuUmVzZXRTdHJlYW1Ub3BpY1Jlc3BvbnNlIgBCQlpAZ2l0aHViLmNvbS9oYXRjaGV0LWRldi9oYXRjaGV0L2ludGVybmFsL3NlcnZpY2VzL3NoYXJlZC9wcm90by92MWIGcHJvdG8z", [file_google_protobuf_timestamp]); /** * @generated from message v1.PublishStreamMessageRequest @@ -230,6 +230,41 @@ export type StreamTopicMetadata = Message<"v1.StreamTopicMetadata"> & { export const StreamTopicMetadataSchema: GenMessage = /*@__PURE__*/ messageDesc(file_v1_streams, 6); +/** + * @generated from message v1.ResetStreamTopicRequest + */ +export type ResetStreamTopicRequest = Message<"v1.ResetStreamTopicRequest"> & { + /** + * @generated from field: string namespace = 1; + */ + namespace: string; + + /** + * @generated from field: string topic = 2; + */ + topic: string; +}; + +/** + * Describes the message v1.ResetStreamTopicRequest. + * Use `create(ResetStreamTopicRequestSchema)` to create a new message. + */ +export const ResetStreamTopicRequestSchema: GenMessage = /*@__PURE__*/ + messageDesc(file_v1_streams, 7); + +/** + * @generated from message v1.ResetStreamTopicResponse + */ +export type ResetStreamTopicResponse = Message<"v1.ResetStreamTopicResponse"> & { +}; + +/** + * Describes the message v1.ResetStreamTopicResponse. + * Use `create(ResetStreamTopicResponseSchema)` to create a new message. + */ +export const ResetStreamTopicResponseSchema: GenMessage = /*@__PURE__*/ + messageDesc(file_v1_streams, 8); + /** * @generated from service v1.V1Streams */ @@ -258,5 +293,13 @@ export const V1Streams: GenService<{ input: typeof GetStreamTopicMetadataRequestSchema; output: typeof StreamTopicMetadataSchema; }, + /** + * @generated from rpc v1.V1Streams.Reset + */ + reset: { + methodKind: "unary"; + input: typeof ResetStreamTopicRequestSchema; + output: typeof ResetStreamTopicResponseSchema; + }, }> = /*@__PURE__*/ serviceDesc(file_v1_streams, 0); diff --git a/sdks/typescript/src/protoc/v1/streams.ts b/sdks/typescript/src/protoc/v1/streams.ts index d1569e0bff..b31d86d5e5 100644 --- a/sdks/typescript/src/protoc/v1/streams.ts +++ b/sdks/typescript/src/protoc/v1/streams.ts @@ -63,6 +63,13 @@ export interface StreamTopicMetadata { lastPublishedAt: Date | undefined; } +export interface ResetStreamTopicRequest { + namespace: string; + topic: string; +} + +export interface ResetStreamTopicResponse {} + function createBasePublishStreamMessageRequest(): PublishStreamMessageRequest { return { namespace: '', @@ -812,6 +819,128 @@ export const StreamTopicMetadata: MessageFns = { }, }; +function createBaseResetStreamTopicRequest(): ResetStreamTopicRequest { + return { namespace: '', topic: '' }; +} + +export const ResetStreamTopicRequest: MessageFns = { + encode( + message: ResetStreamTopicRequest, + writer: BinaryWriter = new BinaryWriter() + ): BinaryWriter { + if (message.namespace !== '') { + writer.uint32(10).string(message.namespace); + } + if (message.topic !== '') { + writer.uint32(18).string(message.topic); + } + return writer; + }, + + decode(input: BinaryReader | Uint8Array, length?: number): ResetStreamTopicRequest { + const reader = input instanceof BinaryReader ? input : new BinaryReader(input); + const end = length === undefined ? reader.len : reader.pos + length; + const message = createBaseResetStreamTopicRequest(); + while (reader.pos < end) { + const tag = reader.uint32(); + switch (tag >>> 3) { + case 1: { + if (tag !== 10) { + break; + } + + message.namespace = reader.string(); + continue; + } + case 2: { + if (tag !== 18) { + break; + } + + message.topic = reader.string(); + continue; + } + } + if ((tag & 7) === 4 || tag === 0) { + break; + } + reader.skip(tag & 7); + } + return message; + }, + + fromJSON(object: any): ResetStreamTopicRequest { + return { + namespace: isSet(object.namespace) ? globalThis.String(object.namespace) : '', + topic: isSet(object.topic) ? globalThis.String(object.topic) : '', + }; + }, + + toJSON(message: ResetStreamTopicRequest): unknown { + const obj: any = {}; + if (message.namespace !== '') { + obj.namespace = message.namespace; + } + if (message.topic !== '') { + obj.topic = message.topic; + } + return obj; + }, + + create(base?: DeepPartial): ResetStreamTopicRequest { + return ResetStreamTopicRequest.fromPartial(base ?? {}); + }, + fromPartial(object: DeepPartial): ResetStreamTopicRequest { + const message = createBaseResetStreamTopicRequest(); + message.namespace = object.namespace ?? ''; + message.topic = object.topic ?? ''; + return message; + }, +}; + +function createBaseResetStreamTopicResponse(): ResetStreamTopicResponse { + return {}; +} + +export const ResetStreamTopicResponse: MessageFns = { + encode(_: ResetStreamTopicResponse, writer: BinaryWriter = new BinaryWriter()): BinaryWriter { + return writer; + }, + + decode(input: BinaryReader | Uint8Array, length?: number): ResetStreamTopicResponse { + const reader = input instanceof BinaryReader ? input : new BinaryReader(input); + const end = length === undefined ? reader.len : reader.pos + length; + const message = createBaseResetStreamTopicResponse(); + while (reader.pos < end) { + const tag = reader.uint32(); + switch (tag >>> 3) { + } + if ((tag & 7) === 4 || tag === 0) { + break; + } + reader.skip(tag & 7); + } + return message; + }, + + fromJSON(_: any): ResetStreamTopicResponse { + return {}; + }, + + toJSON(_: ResetStreamTopicResponse): unknown { + const obj: any = {}; + return obj; + }, + + create(base?: DeepPartial): ResetStreamTopicResponse { + return ResetStreamTopicResponse.fromPartial(base ?? {}); + }, + fromPartial(_: DeepPartial): ResetStreamTopicResponse { + const message = createBaseResetStreamTopicResponse(); + return message; + }, +}; + export type V1StreamsDefinition = typeof V1StreamsDefinition; export const V1StreamsDefinition = { name: 'V1Streams', @@ -841,6 +970,14 @@ export const V1StreamsDefinition = { responseStream: false, options: {}, }, + reset: { + name: 'Reset', + requestType: ResetStreamTopicRequest as typeof ResetStreamTopicRequest, + requestStream: false, + responseType: ResetStreamTopicResponse as typeof ResetStreamTopicResponse, + responseStream: false, + options: {}, + }, }, } as const; @@ -857,6 +994,10 @@ export interface V1StreamsServiceImplementation { request: GetStreamTopicMetadataRequest, context: CallContext & CallContextExt ): Promise>; + reset( + request: ResetStreamTopicRequest, + context: CallContext & CallContextExt + ): Promise>; } export interface V1StreamsClient { @@ -872,6 +1013,10 @@ export interface V1StreamsClient { request: DeepPartial, options?: CallOptions & CallOptionsExt ): Promise; + reset( + request: DeepPartial, + options?: CallOptions & CallOptionsExt + ): Promise; } function bytesFromBase64(b64: string): Uint8Array { diff --git a/sdks/typescript/src/v1/client/features/streams.test.ts b/sdks/typescript/src/v1/client/features/streams.test.ts index 102c16331d..61d659f56b 100644 --- a/sdks/typescript/src/v1/client/features/streams.test.ts +++ b/sdks/typescript/src/v1/client/features/streams.test.ts @@ -318,6 +318,23 @@ describe('StreamsClient.events reconnection', () => { expect(subscribe.mock.calls.map(([req]) => req.cursor)).toEqual([undefined, undefined]); }); + it('resubscribes from the start cursor the server sent, so a reset while disconnected skips nothing', async () => { + const { streams, subscribe } = setup([ + async function* dropped() { + yield { entries: [], hangup: false, cursor: 'start' }; + throw grpcError(Status.INTERNAL); + }, + async function* resumed() { + yield { entries: [], hangup: false, cursor: 'start' }; + yield { entries: [entry(1, 'c1')], hangup: false }; + yield* hangup(); + }, + ]); + + expect(await take(streams.events('topic'), 10)).toEqual([1]); + expect(subscribe.mock.calls.map(([req]) => req.cursor)).toEqual([undefined, 'start']); + }); + it('throws an error a retry cannot fix, such as a cursor expiring while disconnected', async () => { const { streams, subscribe } = setup([ async function* dropped() { @@ -378,6 +395,22 @@ describe('StreamsClient.events reconnection', () => { }); }); +describe('StreamsClient.reset', () => { + it('resets the topic in the default namespace unless one is given', async () => { + const reset = jest.fn().mockResolvedValue({}); + mockedCreateGrpcClient.mockReturnValue({ client: { reset } } as any); + + const streams = new StreamsClient(fakeHatchetClient()); + await streams.reset('topic'); + await streams.reset('topic', { namespace: 'ns' }); + + expect(reset.mock.calls.map(([req]) => req)).toEqual([ + { namespace: '', topic: 'topic' }, + { namespace: 'ns', topic: 'topic' }, + ]); + }); +}); + describe('StreamsClient.topicMetadata', () => { beforeEach(() => { jest.clearAllMocks(); diff --git a/sdks/typescript/src/v1/client/features/streams.ts b/sdks/typescript/src/v1/client/features/streams.ts index 50c4f350cc..c9c0c2fb06 100644 --- a/sdks/typescript/src/v1/client/features/streams.ts +++ b/sdks/typescript/src/v1/client/features/streams.ts @@ -266,7 +266,19 @@ export class StreamsClient { } /** - * Yields topic's messages after options.cursor, or from the oldest retained. + * Makes events() calls without a cursor start after the topic's latest + * message. Readers already iterating, and any call with a cursor, are + * unaffected. + * @param topic - the topic to reset + * @param options - optional namespace override + */ + async reset(topic: string, options?: Pick): Promise { + await this.grpc.reset({ namespace: options?.namespace ?? '', topic }); + } + + /** + * Yields topic's messages after options.cursor, or from the oldest retained + * (or, once the topic is reset, after the reset). * If the connection drops, it resubscribes from the last message yielded, so * nothing is skipped or repeated. Ends when the server hangs up an idle * subscription, options.signal aborts, or the caller stops iterating; throws @@ -289,6 +301,12 @@ export class StreamsClient { return; } + // the first message is where this subscription started, so a + // resubscribe before any entry arrives isn't moved by a reset + if (msg.cursor) { + ({ cursor } = msg); + } + for (const entry of msg.entries) { const payload = entry.payloadRef ? await this.fetchPayload(entry.payloadRef, signal) diff --git a/sdks/typescript/src/v1/examples/durable-streams/durable-streams.e2e.ts b/sdks/typescript/src/v1/examples/durable-streams/durable-streams.e2e.ts index 4e230c47de..5f0255a7f9 100644 --- a/sdks/typescript/src/v1/examples/durable-streams/durable-streams.e2e.ts +++ b/sdks/typescript/src/v1/examples/durable-streams/durable-streams.e2e.ts @@ -66,6 +66,24 @@ describe('durable-streams-e2e', () => { expect(rest.map((e) => decode(e.payload))).toEqual(['three', 'four']); }, 30_000); + it('reset starts new subscribers after the latest message, while existing ones carry on', async () => { + const topic = makeTestScope('durable_streams_reset'); + + await hatchet.streams.publish(topic, 'before-reset'); + + const existing = hatchet.streams.events(topic)[Symbol.asyncIterator](); + expect(decode((await existing.next()).value.payload)).toEqual('before-reset'); + + await hatchet.streams.reset(topic); + await hatchet.streams.publish(topic, 'after-reset'); + + const [fresh] = await collect(hatchet.streams.events(topic), 1); + expect(decode(fresh.payload)).toEqual('after-reset'); + + expect(decode((await existing.next()).value.payload)).toEqual('after-reset'); + await existing.return?.(); + }, 30_000); + it('isolates messages by namespace', async () => { const topic = makeTestScope('durable_streams_namespace'); const namespaceA = 'ns-a'; diff --git a/sdks/typescript/src/v1/examples/durable-streams/run.ts b/sdks/typescript/src/v1/examples/durable-streams/run.ts new file mode 100644 index 0000000000..06d6fc1cef --- /dev/null +++ b/sdks/typescript/src/v1/examples/durable-streams/run.ts @@ -0,0 +1,96 @@ +import { randomUUID } from 'crypto'; +import { hatchet } from '../hatchet-client'; +import { progressTask } from './workflow'; + +const decode = (payload: Uint8Array) => new TextDecoder().decode(payload); + +export async function consume() { + // > Consume + const topic = `progress-${randomUUID()}`; + await progressTask.runNoWait({ topic }); + + for await (const event of hatchet.streams.events(topic)) { + const { step, total } = JSON.parse(decode(event.payload)); + console.log(`step ${step} of ${total}`); + + // the stream outlives the run, so the reader decides when it's done + if (step === total) { + break; + } + } + // !! +} + +export async function publishFromAnywhere() { + // > Publish from anywhere + await hatchet.streams.publish( + 'orders', + JSON.stringify({ orderId: 'ord_123', status: 'shipped' }) + ); + // !! +} + +export async function resumeFromCursor(topic: string) { + // > Resume from a cursor + let cursor: string | undefined; + + for await (const event of hatchet.streams.events(topic)) { + console.log(decode(event.payload)); + ({ cursor } = event); + break; + } + + // picks up right after the last event read, even from another process + for await (const event of hatchet.streams.events(topic, { cursor })) { + console.log(decode(event.payload)); + } + // !! +} + +export async function stopWithSignal(topic: string) { + // > Stop reading + const controller = new AbortController(); + setTimeout(() => controller.abort(), 10_000); + + for await (const event of hatchet.streams.events(topic, { signal: controller.signal })) { + console.log(decode(event.payload)); + } + // !! +} + +export async function namespaces() { + // > Namespaces + await hatchet.streams.publish('updates', 'hello', { namespace: 'customer-a' }); + + for await (const event of hatchet.streams.events('updates', { namespace: 'customer-a' })) { + console.log(decode(event.payload)); + break; + } + // !! +} + +export async function resetTopic() { + // > Reset a topic + await hatchet.streams.reset('orders'); + + // starts after the reset; readers that were already iterating aren't affected + for await (const event of hatchet.streams.events('orders')) { + console.log(decode(event.payload)); + } + // !! +} + +async function main() { + await consume(); + await publishFromAnywhere(); + await stopWithSignal('orders'); + await namespaces(); +} + +if (require.main === module) { + main() + .catch(console.error) + .finally(() => { + process.exit(0); + }); +} diff --git a/sdks/typescript/src/v1/examples/durable-streams/worker.ts b/sdks/typescript/src/v1/examples/durable-streams/worker.ts new file mode 100644 index 0000000000..6379db23d8 --- /dev/null +++ b/sdks/typescript/src/v1/examples/durable-streams/worker.ts @@ -0,0 +1,14 @@ +import { hatchet } from '../hatchet-client'; +import { progressTask } from './workflow'; + +async function main() { + const worker = await hatchet.worker('durable-streams-worker', { + workflows: [progressTask], + }); + + await worker.start(); +} + +if (require.main === module) { + main(); +} diff --git a/sdks/typescript/src/v1/examples/durable-streams/workflow.ts b/sdks/typescript/src/v1/examples/durable-streams/workflow.ts new file mode 100644 index 0000000000..9e62fc04c6 --- /dev/null +++ b/sdks/typescript/src/v1/examples/durable-streams/workflow.ts @@ -0,0 +1,22 @@ +import sleep from '../../../util/sleep'; +import { hatchet } from '../hatchet-client'; + +// > Publishing +export type ProgressInput = { + topic: string; +}; + +export const progressTask = hatchet.task({ + name: 'durable-streams-progress', + fn: async (input: ProgressInput) => { + const total = 5; + + for (let step = 1; step <= total; step += 1) { + await sleep(500); + await hatchet.streams.publish(input.topic, JSON.stringify({ step, total })); + } + + return { steps: total }; + }, +}); +// !! diff --git a/sdks/typescript/src/version.ts b/sdks/typescript/src/version.ts index d7b446e68f..9931e736f9 100644 --- a/sdks/typescript/src/version.ts +++ b/sdks/typescript/src/version.ts @@ -1 +1 @@ -export const HATCHET_VERSION = '1.37.0'; +export const HATCHET_VERSION = '1.38.0'; diff --git a/sql/schema/v1-core.sql b/sql/schema/v1-core.sql index 2af406c717..815ceee40e 100644 --- a/sql/schema/v1-core.sql +++ b/sql/schema/v1-core.sql @@ -2336,6 +2336,8 @@ CREATE TABLE v1_stream_topic ( last_published_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, -- the last offset reserved (see v1_stream_message) last_offset BIGINT NOT NULL DEFAULT 0, + -- subscribers without a cursor start after this offset (see ResetStreamTopic) + reset_offset BIGINT NOT NULL DEFAULT 0, CONSTRAINT v1_stream_topic_pkey PRIMARY KEY (tenant_id, namespace, topic) ) WITH (fillfactor = 80); From 1cd9ed41cee3d74f45fdd3dad1b27d85795f2456 Mon Sep 17 00:00:00 2001 From: Julius Park Date: Tue, 13 Oct 2026 11:05:54 -0400 Subject: [PATCH 2/6] add another test --- examples/typescript/durable-streams/worker.ts | 4 +- .../typescript/durable-streams/workflow.ts | 25 +++++++ .../durable-streams-retry.e2e.ts | 65 +++++++++++++++++++ .../src/v1/examples/durable-streams/worker.ts | 4 +- .../v1/examples/durable-streams/workflow.ts | 26 ++++++++ 5 files changed, 120 insertions(+), 4 deletions(-) create mode 100644 sdks/typescript/src/v1/examples/durable-streams/durable-streams-retry.e2e.ts diff --git a/examples/typescript/durable-streams/worker.ts b/examples/typescript/durable-streams/worker.ts index 6379db23d8..82cc83c7cd 100644 --- a/examples/typescript/durable-streams/worker.ts +++ b/examples/typescript/durable-streams/worker.ts @@ -1,9 +1,9 @@ import { hatchet } from '../hatchet-client'; -import { progressTask } from './workflow'; +import { progressTask, retryingProgressTask } from './workflow'; async function main() { const worker = await hatchet.worker('durable-streams-worker', { - workflows: [progressTask], + workflows: [progressTask, retryingProgressTask], }); await worker.start(); diff --git a/examples/typescript/durable-streams/workflow.ts b/examples/typescript/durable-streams/workflow.ts index 965a7f001f..076f92b8a4 100644 --- a/examples/typescript/durable-streams/workflow.ts +++ b/examples/typescript/durable-streams/workflow.ts @@ -19,3 +19,28 @@ export const progressTask = hatchet.task({ return { steps: total }; }, }); + +// > Reset on retry +export const retryingProgressTask = hatchet.task({ + name: 'durable-streams-retrying-progress', + retries: 1, + fn: async (input: ProgressInput, ctx) => { + // readers that subscribe from here on skip a failed attempt's messages + await hatchet.streams.reset(input.topic); + + for (let chunk = 0; chunk < 3; chunk += 1) { + await hatchet.streams.publish( + input.topic, + JSON.stringify({ attempt: ctx.retryCount(), chunk }) + ); + } + + if (ctx.retryCount() === 0) { + throw new Error('intentional failure before the final message'); + } + + await hatchet.streams.publish(input.topic, JSON.stringify({ done: true })); + + return { chunks: 3 }; + }, +}); diff --git a/sdks/typescript/src/v1/examples/durable-streams/durable-streams-retry.e2e.ts b/sdks/typescript/src/v1/examples/durable-streams/durable-streams-retry.e2e.ts new file mode 100644 index 0000000000..746b0a01da --- /dev/null +++ b/sdks/typescript/src/v1/examples/durable-streams/durable-streams-retry.e2e.ts @@ -0,0 +1,65 @@ +import { makeE2EClient, makeTestScope, startWorker, stopWorker } from '../__e2e__/harness'; +import { retryingProgressTask } from './workflow'; + +type ProgressMessage = { attempt?: number; chunk?: number; done?: boolean; ready?: boolean }; + +const decode = (payload: Uint8Array): ProgressMessage => + JSON.parse(new TextDecoder().decode(payload)); + +// knows only the topic: not the run, its retry count, or where any attempt starts +async function readUntilDone(events: AsyncIterator<{ payload: Uint8Array }>) { + const received: ProgressMessage[] = []; + + for (let next = await events.next(); !next.done; next = await events.next()) { + const msg = decode(next.value.payload); + + if (msg.done) { + await events.return?.(); + break; + } + + if (!msg.ready) { + received.push(msg); + } + } + + return received; +} + +const chunksOf = (attempt: number) => [0, 1, 2].map((chunk) => ({ attempt, chunk })); + +describe('durable-streams-e2e retries', () => { + const hatchet = makeE2EClient(); + let worker: Awaited> | undefined; + + beforeAll(async () => { + worker = await startWorker({ + client: hatchet, + name: 'durable-streams-retry-e2e', + workflows: [retryingProgressTask], + }); + }); + + afterAll(async () => { + await stopWorker(worker); + }); + + it('readers follow a retried task by topic alone, and new readers skip the failed attempt', async () => { + const topic = makeTestScope('durable_streams_retry'); + + // reading the marker back proves the subscription is live before the run's resets + const live = hatchet.streams.events(topic)[Symbol.asyncIterator](); + await hatchet.streams.publish(topic, JSON.stringify({ ready: true })); + expect(decode((await live.next()).value.payload)).toEqual({ ready: true }); + + const liveReading = readUntilDone(live); + + await retryingProgressTask.run({ topic }); + + // a reset doesn't affect a reader already subscribed, so it sees both attempts + expect(await liveReading).toEqual([...chunksOf(0), ...chunksOf(1)]); + + const late = hatchet.streams.events(topic)[Symbol.asyncIterator](); + expect(await readUntilDone(late)).toEqual(chunksOf(1)); + }, 60_000); +}); diff --git a/sdks/typescript/src/v1/examples/durable-streams/worker.ts b/sdks/typescript/src/v1/examples/durable-streams/worker.ts index 6379db23d8..82cc83c7cd 100644 --- a/sdks/typescript/src/v1/examples/durable-streams/worker.ts +++ b/sdks/typescript/src/v1/examples/durable-streams/worker.ts @@ -1,9 +1,9 @@ import { hatchet } from '../hatchet-client'; -import { progressTask } from './workflow'; +import { progressTask, retryingProgressTask } from './workflow'; async function main() { const worker = await hatchet.worker('durable-streams-worker', { - workflows: [progressTask], + workflows: [progressTask, retryingProgressTask], }); await worker.start(); diff --git a/sdks/typescript/src/v1/examples/durable-streams/workflow.ts b/sdks/typescript/src/v1/examples/durable-streams/workflow.ts index 9e62fc04c6..d20e3c6804 100644 --- a/sdks/typescript/src/v1/examples/durable-streams/workflow.ts +++ b/sdks/typescript/src/v1/examples/durable-streams/workflow.ts @@ -20,3 +20,29 @@ export const progressTask = hatchet.task({ }, }); // !! + +// > Reset on retry +export const retryingProgressTask = hatchet.task({ + name: 'durable-streams-retrying-progress', + retries: 1, + fn: async (input: ProgressInput, ctx) => { + // readers that subscribe from here on skip a failed attempt's messages + await hatchet.streams.reset(input.topic); + + for (let chunk = 0; chunk < 3; chunk += 1) { + await hatchet.streams.publish( + input.topic, + JSON.stringify({ attempt: ctx.retryCount(), chunk }) + ); + } + + if (ctx.retryCount() === 0) { + throw new Error('intentional failure before the final message'); + } + + await hatchet.streams.publish(input.topic, JSON.stringify({ done: true })); + + return { chunks: 3 }; + }, +}); +// !! From 117d91d8a5ff828811ff9244126b4604e75383ca Mon Sep 17 00:00:00 2001 From: Julius Park Date: Tue, 13 Oct 2026 11:31:47 -0400 Subject: [PATCH 3/6] clients that reconnect before anything arrives are treated as new subscribers --- internal/services/streams/server.go | 11 ----------- .../src/v1/client/features/streams.test.ts | 17 ----------------- .../src/v1/client/features/streams.ts | 6 ------ 3 files changed, 34 deletions(-) diff --git a/internal/services/streams/server.go b/internal/services/streams/server.go index f9d1735f5a..040ad229eb 100644 --- a/internal/services/streams/server.go +++ b/internal/services/streams/server.go @@ -181,17 +181,6 @@ func (s *ServiceImpl) Subscribe(ctx context.Context, req *contracts.SubscribeStr } } - encodedStart, err := v1.EncodeStreamCursor(cursor) - - if err != nil { - return err - } - - // a client that reconnects before any entry arrives resumes here, not from a later reset - if err := sender.Send(&contracts.StreamMessage{Cursor: encodedStart}); err != nil { - return err - } - ctx, cancel := context.WithCancel(ctx) defer cancel() diff --git a/sdks/typescript/src/v1/client/features/streams.test.ts b/sdks/typescript/src/v1/client/features/streams.test.ts index 61d659f56b..7bbe0e3988 100644 --- a/sdks/typescript/src/v1/client/features/streams.test.ts +++ b/sdks/typescript/src/v1/client/features/streams.test.ts @@ -318,23 +318,6 @@ describe('StreamsClient.events reconnection', () => { expect(subscribe.mock.calls.map(([req]) => req.cursor)).toEqual([undefined, undefined]); }); - it('resubscribes from the start cursor the server sent, so a reset while disconnected skips nothing', async () => { - const { streams, subscribe } = setup([ - async function* dropped() { - yield { entries: [], hangup: false, cursor: 'start' }; - throw grpcError(Status.INTERNAL); - }, - async function* resumed() { - yield { entries: [], hangup: false, cursor: 'start' }; - yield { entries: [entry(1, 'c1')], hangup: false }; - yield* hangup(); - }, - ]); - - expect(await take(streams.events('topic'), 10)).toEqual([1]); - expect(subscribe.mock.calls.map(([req]) => req.cursor)).toEqual([undefined, 'start']); - }); - it('throws an error a retry cannot fix, such as a cursor expiring while disconnected', async () => { const { streams, subscribe } = setup([ async function* dropped() { diff --git a/sdks/typescript/src/v1/client/features/streams.ts b/sdks/typescript/src/v1/client/features/streams.ts index c9c0c2fb06..24d06b7d30 100644 --- a/sdks/typescript/src/v1/client/features/streams.ts +++ b/sdks/typescript/src/v1/client/features/streams.ts @@ -301,12 +301,6 @@ export class StreamsClient { return; } - // the first message is where this subscription started, so a - // resubscribe before any entry arrives isn't moved by a reset - if (msg.cursor) { - ({ cursor } = msg); - } - for (const entry of msg.entries) { const payload = entry.payloadRef ? await this.fetchPayload(entry.payloadRef, signal) From eaa4cfe6203e6cb317b2c1ececd0099d67721c44 Mon Sep 17 00:00:00 2001 From: Julius Park Date: Tue, 13 Oct 2026 12:06:58 -0400 Subject: [PATCH 4/6] fix cosmetic issues --- examples/typescript/durable-streams/workflow.ts | 2 +- examples/typescript/e2e-worker.ts | 3 +++ .../durable-streams/durable-streams-retry.e2e.ts | 15 +-------------- .../src/v1/examples/durable-streams/workflow.ts | 2 +- sdks/typescript/src/v1/examples/e2e-worker.ts | 3 +++ 5 files changed, 9 insertions(+), 16 deletions(-) diff --git a/examples/typescript/durable-streams/workflow.ts b/examples/typescript/durable-streams/workflow.ts index 076f92b8a4..cdb87ecd72 100644 --- a/examples/typescript/durable-streams/workflow.ts +++ b/examples/typescript/durable-streams/workflow.ts @@ -1,4 +1,4 @@ -import sleep from '../../../util/sleep'; +import sleep from '@hatchet-dev/typescript-sdk/util/sleep'; import { hatchet } from '../hatchet-client'; // > Publishing diff --git a/examples/typescript/e2e-worker.ts b/examples/typescript/e2e-worker.ts index ebdf12de72..78ae51d732 100644 --- a/examples/typescript/e2e-worker.ts +++ b/examples/typescript/e2e-worker.ts @@ -57,6 +57,7 @@ import { callbackOrderingRoot, } from './durable_callback_ordering/workflow'; import { durableSleep } from './durable_sleep/workflow'; +import { progressTask, retryingProgressTask } from './durable-streams/workflow'; import { createLoggingWorkflow } from './logger/workflow'; import { nonRetryableWorkflow } from './non_retryable/workflow'; import { retryAfterExponentialBackoff, retryAfterUpstreamDelay } from './retry_after/workflow'; @@ -140,6 +141,8 @@ const workflows = [ callbackOrderingMid, callbackOrderingRoot, durableSleep, + progressTask, + retryingProgressTask, evictableSleep, evictableWaitForEvent, evictableMemoThenWaitForEvent, diff --git a/sdks/typescript/src/v1/examples/durable-streams/durable-streams-retry.e2e.ts b/sdks/typescript/src/v1/examples/durable-streams/durable-streams-retry.e2e.ts index 746b0a01da..12d76aee35 100644 --- a/sdks/typescript/src/v1/examples/durable-streams/durable-streams-retry.e2e.ts +++ b/sdks/typescript/src/v1/examples/durable-streams/durable-streams-retry.e2e.ts @@ -1,4 +1,4 @@ -import { makeE2EClient, makeTestScope, startWorker, stopWorker } from '../__e2e__/harness'; +import { makeE2EClient, makeTestScope } from '../__e2e__/harness'; import { retryingProgressTask } from './workflow'; type ProgressMessage = { attempt?: number; chunk?: number; done?: boolean; ready?: boolean }; @@ -30,19 +30,6 @@ const chunksOf = (attempt: number) => [0, 1, 2].map((chunk) => ({ attempt, chunk describe('durable-streams-e2e retries', () => { const hatchet = makeE2EClient(); - let worker: Awaited> | undefined; - - beforeAll(async () => { - worker = await startWorker({ - client: hatchet, - name: 'durable-streams-retry-e2e', - workflows: [retryingProgressTask], - }); - }); - - afterAll(async () => { - await stopWorker(worker); - }); it('readers follow a retried task by topic alone, and new readers skip the failed attempt', async () => { const topic = makeTestScope('durable_streams_retry'); diff --git a/sdks/typescript/src/v1/examples/durable-streams/workflow.ts b/sdks/typescript/src/v1/examples/durable-streams/workflow.ts index d20e3c6804..89e6803c49 100644 --- a/sdks/typescript/src/v1/examples/durable-streams/workflow.ts +++ b/sdks/typescript/src/v1/examples/durable-streams/workflow.ts @@ -1,4 +1,4 @@ -import sleep from '../../../util/sleep'; +import sleep from '@hatchet/util/sleep'; import { hatchet } from '../hatchet-client'; // > Publishing diff --git a/sdks/typescript/src/v1/examples/e2e-worker.ts b/sdks/typescript/src/v1/examples/e2e-worker.ts index ebdf12de72..78ae51d732 100644 --- a/sdks/typescript/src/v1/examples/e2e-worker.ts +++ b/sdks/typescript/src/v1/examples/e2e-worker.ts @@ -57,6 +57,7 @@ import { callbackOrderingRoot, } from './durable_callback_ordering/workflow'; import { durableSleep } from './durable_sleep/workflow'; +import { progressTask, retryingProgressTask } from './durable-streams/workflow'; import { createLoggingWorkflow } from './logger/workflow'; import { nonRetryableWorkflow } from './non_retryable/workflow'; import { retryAfterExponentialBackoff, retryAfterUpstreamDelay } from './retry_after/workflow'; @@ -140,6 +141,8 @@ const workflows = [ callbackOrderingMid, callbackOrderingRoot, durableSleep, + progressTask, + retryingProgressTask, evictableSleep, evictableWaitForEvent, evictableMemoThenWaitForEvent, From 197986b6b95f3cb357020867d28c4959ac295e67 Mon Sep 17 00:00:00 2001 From: Julius Park Date: Tue, 13 Oct 2026 12:37:20 -0400 Subject: [PATCH 5/6] add godoc comments --- internal/services/streams/server.go | 2 ++ pkg/repository/streams.go | 4 ++++ 2 files changed, 6 insertions(+) diff --git a/internal/services/streams/server.go b/internal/services/streams/server.go index 040ad229eb..4aab13ef74 100644 --- a/internal/services/streams/server.go +++ b/internal/services/streams/server.go @@ -216,6 +216,8 @@ func (s *ServiceImpl) Subscribe(ctx context.Context, req *contracts.SubscribeStr return nil } +// Reset makes subscribers without a cursor start after the topic's latest +// message. Subscriptions already open, and any with a cursor, are unaffected. func (s *ServiceImpl) Reset(ctx context.Context, req *contracts.ResetStreamTopicRequest) (*contracts.ResetStreamTopicResponse, error) { tenant := ctx.Value("tenant").(*sqlcv1.Tenant) tenantId := tenant.ID diff --git a/pkg/repository/streams.go b/pkg/repository/streams.go index b1fb296837..a2226ea606 100644 --- a/pkg/repository/streams.go +++ b/pkg/repository/streams.go @@ -599,6 +599,8 @@ func (r *streamsRepositoryImpl) CheckCursorRetained(ctx context.Context, tenantI return nil } +// ResetTopic sets the topic's reset offset to its last reserved offset. It waits +// on in-flight publishes' row lock, so every message up to that offset is stored. func (r *streamsRepositoryImpl) ResetTopic(ctx context.Context, tenantId uuid.UUID, namespace, topic string) error { return r.queries.ResetStreamTopic(ctx, r.pool, sqlcv1.ResetStreamTopicParams{ Tenantid: tenantId, @@ -607,6 +609,8 @@ func (r *streamsRepositoryImpl) ResetTopic(ctx context.Context, tenantId uuid.UU }) } +// TopicStartCursor returns a cursor at the topic's reset offset, or its start for +// a topic never reset or published to. It has no time, so it never expires. func (r *streamsRepositoryImpl) TopicStartCursor(ctx context.Context, tenantId uuid.UUID, namespace, topic string) (StreamCursor, error) { resetOffset, err := r.queries.GetStreamTopicResetOffset(ctx, r.pool, sqlcv1.GetStreamTopicResetOffsetParams{ Tenantid: tenantId, From 9809a9691106e51e31d02354b3c2bed7dc0d9f99 Mon Sep 17 00:00:00 2001 From: Julius Park Date: Tue, 13 Oct 2026 12:47:06 -0400 Subject: [PATCH 6/6] simplify and rename --- .../migrations/20261013133204_v1_0_172.sql | 4 +-- pkg/repository/sqlcv1/models.go | 2 +- pkg/repository/sqlcv1/streams.sql | 13 +++++---- pkg/repository/sqlcv1/streams.sql.go | 29 ++++++++++--------- pkg/repository/streams.go | 13 ++++----- sql/schema/v1-core.sql | 2 +- 6 files changed, 34 insertions(+), 29 deletions(-) diff --git a/cmd/hatchet-migrate/migrate/migrations/20261013133204_v1_0_172.sql b/cmd/hatchet-migrate/migrate/migrations/20261013133204_v1_0_172.sql index fe4601e620..45a8a980dc 100644 --- a/cmd/hatchet-migrate/migrate/migrations/20261013133204_v1_0_172.sql +++ b/cmd/hatchet-migrate/migrate/migrations/20261013133204_v1_0_172.sql @@ -1,10 +1,10 @@ -- +goose Up -- +goose StatementBegin -- subscribers without a cursor start after this offset (see ResetStreamTopic) -ALTER TABLE v1_stream_topic ADD COLUMN reset_offset BIGINT NOT NULL DEFAULT 0; +ALTER TABLE v1_stream_topic ADD COLUMN starting_offset BIGINT NOT NULL DEFAULT 0; -- +goose StatementEnd -- +goose Down -- +goose StatementBegin -ALTER TABLE v1_stream_topic DROP COLUMN reset_offset; +ALTER TABLE v1_stream_topic DROP COLUMN starting_offset; -- +goose StatementEnd diff --git a/pkg/repository/sqlcv1/models.go b/pkg/repository/sqlcv1/models.go index 3cd3fc62f5..16345ee7bf 100644 --- a/pkg/repository/sqlcv1/models.go +++ b/pkg/repository/sqlcv1/models.go @@ -3888,7 +3888,7 @@ type V1StreamTopic struct { InsertedAt pgtype.Timestamptz `json:"inserted_at"` LastPublishedAt pgtype.Timestamptz `json:"last_published_at"` LastOffset int64 `json:"last_offset"` - ResetOffset int64 `json:"reset_offset"` + StartingOffset int64 `json:"starting_offset"` } type V1Task struct { diff --git a/pkg/repository/sqlcv1/streams.sql b/pkg/repository/sqlcv1/streams.sql index 81e61785f5..f12c022c68 100644 --- a/pkg/repository/sqlcv1/streams.sql +++ b/pkg/repository/sqlcv1/streams.sql @@ -278,10 +278,13 @@ WHERE EXISTS (SELECT 1 FROM topic); -- Waits on the row lock a publish holds until commit (see -- ReserveStreamTopicOffsets), so every offset up to last_offset is stored. UPDATE v1_stream_topic -SET reset_offset = last_offset +SET starting_offset = last_offset WHERE tenant_id = @tenantId::uuid AND namespace = @namespace::text AND topic = @topic::text; --- name: GetStreamTopicResetOffset :one -SELECT reset_offset -FROM v1_stream_topic -WHERE tenant_id = @tenantId::uuid AND namespace = @namespace::text AND topic = @topic::text; +-- name: GetStreamTopicStartingOffset :one +-- 0 for a topic not yet published to, so it starts at its beginning. +SELECT COALESCE(( + SELECT starting_offset + FROM v1_stream_topic + WHERE tenant_id = @tenantId::uuid AND namespace = @namespace::text AND topic = @topic::text +), 0)::bigint AS starting_offset; diff --git a/pkg/repository/sqlcv1/streams.sql.go b/pkg/repository/sqlcv1/streams.sql.go index 05c16079cf..ff79748aa4 100644 --- a/pkg/repository/sqlcv1/streams.sql.go +++ b/pkg/repository/sqlcv1/streams.sql.go @@ -282,23 +282,26 @@ func (q *Queries) GetStreamTopicMetadata(ctx context.Context, db DBTX, arg GetSt return &i, err } -const getStreamTopicResetOffset = `-- name: GetStreamTopicResetOffset :one -SELECT reset_offset -FROM v1_stream_topic -WHERE tenant_id = $1::uuid AND namespace = $2::text AND topic = $3::text +const getStreamTopicStartingOffset = `-- name: GetStreamTopicStartingOffset :one +SELECT COALESCE(( + SELECT starting_offset + FROM v1_stream_topic + WHERE tenant_id = $1::uuid AND namespace = $2::text AND topic = $3::text +), 0)::bigint AS starting_offset ` -type GetStreamTopicResetOffsetParams struct { +type GetStreamTopicStartingOffsetParams struct { Tenantid uuid.UUID `json:"tenantid"` Namespace string `json:"namespace"` Topic string `json:"topic"` } -func (q *Queries) GetStreamTopicResetOffset(ctx context.Context, db DBTX, arg GetStreamTopicResetOffsetParams) (int64, error) { - row := db.QueryRow(ctx, getStreamTopicResetOffset, arg.Tenantid, arg.Namespace, arg.Topic) - var reset_offset int64 - err := row.Scan(&reset_offset) - return reset_offset, err +// 0 for a topic not yet published to, so it starts at its beginning. +func (q *Queries) GetStreamTopicStartingOffset(ctx context.Context, db DBTX, arg GetStreamTopicStartingOffsetParams) (int64, error) { + row := db.QueryRow(ctx, getStreamTopicStartingOffset, arg.Tenantid, arg.Namespace, arg.Topic) + var starting_offset int64 + err := row.Scan(&starting_offset) + return starting_offset, err } const insertStreamPayload = `-- name: InsertStreamPayload :one @@ -543,7 +546,7 @@ func (q *Queries) ListStreamRetentionDeleteCandidates(ctx context.Context, db DB const resetStreamTopic = `-- name: ResetStreamTopic :exec UPDATE v1_stream_topic -SET reset_offset = last_offset +SET starting_offset = last_offset WHERE tenant_id = $1::uuid AND namespace = $2::text AND topic = $3::text ` @@ -619,7 +622,7 @@ INSERT INTO v1_stream_topic (tenant_id, namespace, topic) VALUES ($1::uuid, $2::text, $3::text) ON CONFLICT (tenant_id, namespace, topic) DO UPDATE SET last_published_at = NOW() -RETURNING v1_stream_topic.tenant_id, v1_stream_topic.namespace, v1_stream_topic.topic, v1_stream_topic.inserted_at, v1_stream_topic.last_published_at, v1_stream_topic.last_offset, v1_stream_topic.reset_offset, (xmax = 0) AS inserted +RETURNING v1_stream_topic.tenant_id, v1_stream_topic.namespace, v1_stream_topic.topic, v1_stream_topic.inserted_at, v1_stream_topic.last_published_at, v1_stream_topic.last_offset, v1_stream_topic.starting_offset, (xmax = 0) AS inserted ` type UpsertStreamTopicParams struct { @@ -644,7 +647,7 @@ func (q *Queries) UpsertStreamTopic(ctx context.Context, db DBTX, arg UpsertStre &i.V1StreamTopic.InsertedAt, &i.V1StreamTopic.LastPublishedAt, &i.V1StreamTopic.LastOffset, - &i.V1StreamTopic.ResetOffset, + &i.V1StreamTopic.StartingOffset, &i.Inserted, ) return &i, err diff --git a/pkg/repository/streams.go b/pkg/repository/streams.go index a2226ea606..e56d9afa41 100644 --- a/pkg/repository/streams.go +++ b/pkg/repository/streams.go @@ -599,7 +599,7 @@ func (r *streamsRepositoryImpl) CheckCursorRetained(ctx context.Context, tenantI return nil } -// ResetTopic sets the topic's reset offset to its last reserved offset. It waits +// ResetTopic sets the topic's starting offset to its last reserved offset. It waits // on in-flight publishes' row lock, so every message up to that offset is stored. func (r *streamsRepositoryImpl) ResetTopic(ctx context.Context, tenantId uuid.UUID, namespace, topic string) error { return r.queries.ResetStreamTopic(ctx, r.pool, sqlcv1.ResetStreamTopicParams{ @@ -609,19 +609,18 @@ func (r *streamsRepositoryImpl) ResetTopic(ctx context.Context, tenantId uuid.UU }) } -// TopicStartCursor returns a cursor at the topic's reset offset, or its start for -// a topic never reset or published to. It has no time, so it never expires. +// TopicStartCursor returns a cursor at the topic's starting offset, 0 until it's +// reset. It has no time, so it never expires. func (r *streamsRepositoryImpl) TopicStartCursor(ctx context.Context, tenantId uuid.UUID, namespace, topic string) (StreamCursor, error) { - resetOffset, err := r.queries.GetStreamTopicResetOffset(ctx, r.pool, sqlcv1.GetStreamTopicResetOffsetParams{ + startingOffset, err := r.queries.GetStreamTopicStartingOffset(ctx, r.pool, sqlcv1.GetStreamTopicStartingOffsetParams{ Tenantid: tenantId, Namespace: namespace, Topic: topic, }) - if err != nil && !errors.Is(err, pgx.ErrNoRows) { + if err != nil { return StreamCursor{}, err } - // offsets start at 1, so ID 0 is the start of the topic - return StreamCursor{Namespace: namespace, Topic: topic, ID: resetOffset}, nil + return StreamCursor{Namespace: namespace, Topic: topic, ID: startingOffset}, nil } diff --git a/sql/schema/v1-core.sql b/sql/schema/v1-core.sql index 815ceee40e..b58df392e2 100644 --- a/sql/schema/v1-core.sql +++ b/sql/schema/v1-core.sql @@ -2337,7 +2337,7 @@ CREATE TABLE v1_stream_topic ( -- the last offset reserved (see v1_stream_message) last_offset BIGINT NOT NULL DEFAULT 0, -- subscribers without a cursor start after this offset (see ResetStreamTopic) - reset_offset BIGINT NOT NULL DEFAULT 0, + starting_offset BIGINT NOT NULL DEFAULT 0, CONSTRAINT v1_stream_topic_pkey PRIMARY KEY (tenant_id, namespace, topic) ) WITH (fillfactor = 80);