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..45a8a980dc --- /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 starting_offset BIGINT NOT NULL DEFAULT 0; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +ALTER TABLE v1_stream_topic DROP COLUMN starting_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..82cc83c7cd --- /dev/null +++ b/examples/typescript/durable-streams/worker.ts @@ -0,0 +1,14 @@ +import { hatchet } from '../hatchet-client'; +import { progressTask, retryingProgressTask } from './workflow'; + +async function main() { + const worker = await hatchet.worker('durable-streams-worker', { + workflows: [progressTask, retryingProgressTask], + }); + + 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..cdb87ecd72 --- /dev/null +++ b/examples/typescript/durable-streams/workflow.ts @@ -0,0 +1,46 @@ +import sleep from '@hatchet-dev/typescript-sdk/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 }; + }, +}); + +// > 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/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/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..4aab13ef74 100644 --- a/internal/services/streams/server.go +++ b/internal/services/streams/server.go @@ -175,6 +175,12 @@ 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 + } + } + ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -210,6 +216,39 @@ 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 + + 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..16345ee7bf 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"` + StartingOffset int64 `json:"starting_offset"` } type V1Task struct { diff --git a/pkg/repository/sqlcv1/streams.sql b/pkg/repository/sqlcv1/streams.sql index 5dc567b0fc..f12c022c68 100644 --- a/pkg/repository/sqlcv1/streams.sql +++ b/pkg/repository/sqlcv1/streams.sql @@ -273,3 +273,18 @@ 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 starting_offset = last_offset +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 2560f6977b..ff79748aa4 100644 --- a/pkg/repository/sqlcv1/streams.sql.go +++ b/pkg/repository/sqlcv1/streams.sql.go @@ -282,6 +282,28 @@ func (q *Queries) GetStreamTopicMetadata(ctx context.Context, db DBTX, arg GetSt return &i, err } +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 GetStreamTopicStartingOffsetParams struct { + Tenantid uuid.UUID `json:"tenantid"` + Namespace string `json:"namespace"` + Topic string `json:"topic"` +} + +// 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 INSERT INTO v1_stream_payload (id, tenant_id, payload) VALUES ($1::uuid, $2::uuid, $3::bytea) @@ -522,6 +544,25 @@ func (q *Queries) ListStreamRetentionDeleteCandidates(ctx context.Context, db DB return items, nil } +const resetStreamTopic = `-- name: ResetStreamTopic :exec +UPDATE v1_stream_topic +SET starting_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 +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, (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 { @@ -606,6 +647,7 @@ func (q *Queries) UpsertStreamTopic(ctx context.Context, db DBTX, arg UpsertStre &i.V1StreamTopic.InsertedAt, &i.V1StreamTopic.LastPublishedAt, &i.V1StreamTopic.LastOffset, + &i.V1StreamTopic.StartingOffset, &i.Inserted, ) return &i, err diff --git a/pkg/repository/streams.go b/pkg/repository/streams.go index c0211470fd..e56d9afa41 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,29 @@ func (r *streamsRepositoryImpl) CheckCursorRetained(ctx context.Context, tenantI return nil } + +// 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{ + Tenantid: tenantId, + Namespace: namespace, + Topic: topic, + }) +} + +// 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) { + startingOffset, err := r.queries.GetStreamTopicStartingOffset(ctx, r.pool, sqlcv1.GetStreamTopicStartingOffsetParams{ + Tenantid: tenantId, + Namespace: namespace, + Topic: topic, + }) + + if err != nil { + return StreamCursor{}, err + } + + return StreamCursor{Namespace: namespace, Topic: topic, ID: startingOffset}, 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..7bbe0e3988 100644 --- a/sdks/typescript/src/v1/client/features/streams.test.ts +++ b/sdks/typescript/src/v1/client/features/streams.test.ts @@ -378,6 +378,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..24d06b7d30 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 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..12d76aee35 --- /dev/null +++ b/sdks/typescript/src/v1/examples/durable-streams/durable-streams-retry.e2e.ts @@ -0,0 +1,52 @@ +import { makeE2EClient, makeTestScope } 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(); + + 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/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..82cc83c7cd --- /dev/null +++ b/sdks/typescript/src/v1/examples/durable-streams/worker.ts @@ -0,0 +1,14 @@ +import { hatchet } from '../hatchet-client'; +import { progressTask, retryingProgressTask } from './workflow'; + +async function main() { + const worker = await hatchet.worker('durable-streams-worker', { + workflows: [progressTask, retryingProgressTask], + }); + + 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..89e6803c49 --- /dev/null +++ b/sdks/typescript/src/v1/examples/durable-streams/workflow.ts @@ -0,0 +1,48 @@ +import sleep from '@hatchet/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 }; + }, +}); +// !! + +// > 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/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, 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..b58df392e2 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) + starting_offset BIGINT NOT NULL DEFAULT 0, CONSTRAINT v1_stream_topic_pkey PRIMARY KEY (tenant_id, namespace, topic) ) WITH (fillfactor = 80);