refactor: separate workload class from placement driver

internal/assignmentqueue/jetstream.go

@@ -19,7 +19,7 @@ type publishAPI interface {
 	PublishMsg(context.Context, *nats.Msg, ...jetstream.PublishOpt) (*jetstream.PubAck, error)
 }
 
-// Publisher implements the scheduler dispatcher with one subject per backend.
+// Publisher implements the scheduler dispatcher with one subject per placement.
 type Publisher struct {
 	JetStream   publishAPI
 	SubjectBase string
@@ -38,7 +38,7 @@ func (p Publisher) Dispatch(ctx context.Context, assignment taskassignment.Assig
 		return errors.New("assignment subject base is required")
 	}
 	message := &nats.Msg{
-		Subject: base + "." + string(assignment.Backend),
+		Subject: base + "." + assignment.Placement.Key(),
 		Header:  nats.Header{jetstream.MsgIDHeader: []string{assignment.ID}},
 		Data:    body,
 	}
@@ -86,13 +86,13 @@ type Processor struct {
 type Event struct {
 	Name         string
 	AssignmentID string
-	Backend      taskassignment.Backend
+	Placement    taskassignment.Placement
 	RetryDelay   time.Duration
 }
 
 func (p Processor) event(name string, assignment taskassignment.Assignment, retryDelay time.Duration) {
 	if p.OnEvent != nil {
-		p.OnEvent(Event{Name: name, AssignmentID: assignment.ID, Backend: assignment.Backend, RetryDelay: retryDelay})
+		p.OnEvent(Event{Name: name, AssignmentID: assignment.ID, Placement: assignment.Placement, RetryDelay: retryDelay})
 	}
 }
 
@@ -178,24 +178,25 @@ type consumerManager interface {
 
 // OpenConsumer creates the durable backend cursor. Capacity is enforced both
 // server-side and by ConsumerComponent's local semaphore.
-func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectBase string, backend taskassignment.Backend, capacity int) (jetstream.Consumer, error) {
+func OpenConsumer(ctx context.Context, manager consumerManager, stream, subjectBase string, placement taskassignment.Placement, capacity int) (jetstream.Consumer, error) {
 	if manager == nil || stream == "" || strings.TrimSuffix(subjectBase, ".") == "" || capacity < 1 {
 		return nil, errors.New("JetStream manager, stream, subject base, and positive capacity are required")
 	}
-	if backend != taskassignment.BackendPod && backend != taskassignment.BackendVM {
-		return nil, fmt.Errorf("unsupported assignment backend %q", backend)
+	if err := placement.Validate(); err != nil {
+		return nil, err
 	}
+	key := placement.Key()
 	consumer, err := manager.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{
-		Name:          string(backend),
-		Durable:       string(backend),
-		FilterSubject: strings.TrimSuffix(subjectBase, ".") + "." + string(backend),
+		Name:          key,
+		Durable:       key,
+		FilterSubject: strings.TrimSuffix(subjectBase, ".") + "." + key,
 		AckPolicy:     jetstream.AckExplicitPolicy,
 		AckWait:       5 * time.Minute,
 		MaxAckPending: capacity,
 		MaxDeliver:    1000,
 	})
 	if err != nil {
-		return nil, fmt.Errorf("open %s assignment consumer: %w", backend, err)
+		return nil, fmt.Errorf("open %s assignment consumer: %w", key, err)
 	}
 	return consumer, nil
 }

internal/assignmentqueue/jetstream_test.go

@@ -38,13 +38,13 @@ func (p *fakePublisher) PublishMsg(_ context.Context, message *nats.Msg, _ ...je
 	return &jetstream.PubAck{}, nil
 }
 
-func TestPublisherUsesBackendSubjectAndAssignmentDeduplication(t *testing.T) {
+func TestPublisherUsesPlacementSubjectAndAssignmentDeduplication(t *testing.T) {
 	api := &fakePublisher{}
 	publisher := Publisher{JetStream: api, SubjectBase: "ci.assignment"}
 	if err := publisher.Dispatch(context.Background(), testAssignment(t)); err != nil {
 		t.Fatal(err)
 	}
-	if api.message.Subject != "ci.assignment.pod" {
+	if api.message.Subject != "ci.assignment.container.kubernetes" {
 		t.Fatalf("subject = %q", api.message.Subject)
 	}
 	if api.message.Header.Get(jetstream.MsgIDHeader) != "gitea-task-42" {
@@ -139,7 +139,7 @@ func TestProcessorAcknowledgesPersistedHandoff(t *testing.T) {
 		t.Fatalf("events = %#v", events)
 	}
 	for index := range want {
-		if events[index].Name != want[index] || events[index].AssignmentID != "gitea-task-42" || events[index].Backend != taskassignment.BackendPod {
+		if events[index].Name != want[index] || events[index].AssignmentID != "gitea-task-42" || events[index].Placement != taskassignment.KubernetesContainer {
 			t.Fatalf("event[%d] = %#v", index, events[index])
 		}
 	}
@@ -210,12 +210,12 @@ func (m *fakeConsumerManager) CreateOrUpdateConsumer(_ context.Context, _ string
 	return nil, nil
 }
 
-func TestOpenConsumerUsesIndependentDurablePerBackend(t *testing.T) {
+func TestOpenConsumerUsesIndependentDurablePerPlacement(t *testing.T) {
 	manager := &fakeConsumerManager{}
-	if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.BackendPod, 4); err != nil {
+	if _, err := OpenConsumer(context.Background(), manager, "CI_RUNNER", "ci.assignment", taskassignment.KubernetesContainer, 4); err != nil {
 		t.Fatal(err)
 	}
-	if manager.config.Durable != "pod" || manager.config.FilterSubject != "ci.assignment.pod" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 || manager.config.MaxDeliver != 1000 {
+	if manager.config.Durable != "container.kubernetes" || manager.config.FilterSubject != "ci.assignment.container.kubernetes" || manager.config.AckPolicy != jetstream.AckExplicitPolicy || manager.config.MaxAckPending != 4 || manager.config.MaxDeliver != 1000 {
 		t.Fatalf("config = %#v", manager.config)
 	}
 }