Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,11 @@
# Changelog

## v0.3.4 — 2026-05-13

### Fixed

- **Include `google/protobuf/duration.proto` in `ContractRegistry.FileDescriptorSet`.** `eventbus.proto` imports `duration.proto` for `StreamConfig.max_age` and `ConsumerConfig.ack_wait`. `protodesc.NewFiles` requires every imported file to be present in the FileDescriptorSet, so the v0.3.3 set (struct + empty + eventbus) silently failed to register any eventbus messages — the engine's contract-types resolver ended up empty, and STRICT_PROTO modules fell through to `protoregistry.GlobalTypes` (which still doesn't contain plugin-local protos). Result: PR 279 image-launch *still* hit `generated codec ... not found` after the v0.3.3 bump. v0.3.4 adds the duration well-known type before the eventbus file in the descriptor set so resolution completes.

## v0.3.3 — 2026-05-13

### Fixed
Expand Down
6 changes: 6 additions & 0 deletions cmd/workflow-plugin-eventbus/plugin.go
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import (
"google.golang.org/protobuf/reflect/protodesc"
"google.golang.org/protobuf/types/descriptorpb"
"google.golang.org/protobuf/types/known/anypb"
"google.golang.org/protobuf/types/known/durationpb"
"google.golang.org/protobuf/types/known/emptypb"
"google.golang.org/protobuf/types/known/structpb"

Expand Down Expand Up @@ -228,6 +229,11 @@ func (p *eventbusPlugin) ContractRegistry() *pb.ContractRegistry {
File: []*descriptorpb.FileDescriptorProto{
protodesc.ToFileDescriptorProto(structpb.File_google_protobuf_struct_proto),
protodesc.ToFileDescriptorProto(emptypb.File_google_protobuf_empty_proto),
// eventbus.proto imports google/protobuf/duration.proto for
// StreamConfig.max_age + ConsumerConfig.ack_wait — protodesc.NewFiles
// requires every imported file to be present in the FileDescriptorSet,
// so include the duration well-known type BEFORE eventbus_proto.
Comment on lines +233 to +235
protodesc.ToFileDescriptorProto(durationpb.File_google_protobuf_duration_proto),
protodesc.ToFileDescriptorProto(eventbusv1.File_eventbus_proto),
},
},
Expand Down
67 changes: 67 additions & 0 deletions cmd/workflow-plugin-eventbus/plugin_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,77 @@ import (
"strings"
"testing"

"google.golang.org/protobuf/reflect/protodesc"
"google.golang.org/protobuf/reflect/protoreflect"
"google.golang.org/protobuf/reflect/protoregistry"
"google.golang.org/protobuf/types/dynamicpb"

eventbus "github.com/GoCodeAlone/workflow-plugin-eventbus"
eventbusv1 "github.com/GoCodeAlone/workflow-plugin-eventbus/gen"
)

// TestContractRegistry_FileDescriptorSetResolvesEveryContractMessage is the
// regression test for v0.3.3's incomplete FileDescriptorSet (missing
// google/protobuf/duration.proto which eventbus.proto imports). The engine
// builds its contract-types resolver via protodesc.NewFiles(registry.
// FileDescriptorSet). If any imported file is missing, NewFiles silently
// produces a registry that never registers the eventbus messages, and the
// engine falls through to protoregistry.GlobalTypes (which does not contain
// plugin-local protos) at module construction time — failing with
// "generated codec for protobuf message X not found: proto: not found".
Comment on lines +20 to +24
Comment on lines +18 to +24
//
// This test exercises the exact engine path: read the plugin's
// ContractRegistry, register the FileDescriptorSet, then assert that every
// Config/Input/Output/Trigger message named in Contracts can be resolved.
func TestContractRegistry_FileDescriptorSetResolvesEveryContractMessage(t *testing.T) {
p := &eventbusPlugin{}
reg := p.ContractRegistry()
if reg == nil {
t.Fatal("ContractRegistry returned nil")
}
if reg.FileDescriptorSet == nil || len(reg.FileDescriptorSet.File) == 0 {
t.Fatal("ContractRegistry must populate FileDescriptorSet so the engine can resolve plugin-local proto codecs")
}
files, err := protodesc.NewFiles(reg.FileDescriptorSet)
if err != nil {
t.Fatalf("protodesc.NewFiles failed — every imported file must be present in FileDescriptorSet: %v", err)
}
types := new(protoregistry.Types)
files.RangeFiles(func(file protoreflect.FileDescriptor) bool {
registerMessages(t, types, file.Messages())
return true
})
// Every distinct message name referenced in Contracts (Config/Input/Output)
// must be resolvable. This mirrors the workflow engine's
// `newMessageByName(messageName, contractTypes)` lookup at
// createTypedConfigRequest time.
wantNames := map[string]struct{}{}
for _, c := range reg.Contracts {
for _, name := range []string{c.ConfigMessage, c.InputMessage, c.OutputMessage} {
if name == "" {
continue
}
wantNames[name] = struct{}{}
}
}
for name := range wantNames {
if _, err := types.FindMessageByName(protoreflect.FullName(name)); err != nil {
t.Errorf("FileDescriptorSet missing codec for contract message %q: %v", name, err)
}
}
}

func registerMessages(t *testing.T, types *protoregistry.Types, messages protoreflect.MessageDescriptors) {
t.Helper()
for i := 0; i < messages.Len(); i++ {
m := messages.Get(i)
if err := types.RegisterMessage(dynamicpb.NewMessageType(m)); err != nil {
t.Fatalf("register %q: %v", m.FullName(), err)
}
registerMessages(t, types, m.Messages())
}
}

// TestCreateTrigger_AliasConsumerToName verifies that BMW-style configs supplying
// `consumer` instead of the proto-canonical `name` build successfully. This is
// the core fix shipped in v0.3.1 — without this, BMW pipelines fail with
Expand Down
2 changes: 1 addition & 1 deletion plugin.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "workflow-plugin-eventbus",
"version": "0.3.3",
"version": "0.3.4",
"description": "Provisions durable event-bus clusters (NATS / Kafka / Kinesis) as IaC and exposes typed pipeline steps for publish / consume operations.",
"author": "GoCodeAlone",
"license": "MIT",
Expand Down
Loading