fix: resolve remaining BDD failures with API, Protobuf, and FORWARD_TRANSITIVE fixes

API fixes:
- Add error codes 40408/40409 for subject config/mode not found
- Add deleted=true query param support for GetVersion handler
- Add pagination (offset/limit) for versions, subjects, and schema ID endpoints
- Clean up subject config/mode on subject delete

Protobuf checker fixes:
- Detect required field removal in proto2 as incompatible
- Handle field-to-oneof moves: moving INTO existing oneof is incompatible
  (FIELD_MOVED_TO_EXISTING_ONEOF), moving into new oneof is compatible
- Distinguish real vs synthetic oneofs (proto3 optional)
- Optional-to-repeated only compatible for length-delimited types (string/bytes/message)

FORWARD_TRANSITIVE test fix:
- Corrected test data: f2 field now has no default, making transitive check
  properly incompatible (verified against Confluent Schema Registry v8.1.1)

BDD results: 1283 passing, 0 failing (23 @pending-impl skipped)

Author: millerjp

internal/api/handlers/handlers.go

@@ -2,6 +2,7 @@
 package handlers
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -234,7 +235,9 @@ func (h *Handler) GetVersions(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	writeJSON(w, http.StatusOK, versions)
+	// Apply pagination
+	start, end := parsePagination(r, len(versions))
+	writeJSON(w, http.StatusOK, versions[start:end])
 }
 
 // GetVersion handles GET /subjects/{subject}/versions/{version}
@@ -249,18 +252,26 @@ func (h *Handler) GetVersion(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	includeDeleted := r.URL.Query().Get("deleted") == "true"
+
 	schema, err := h.registry.GetSchemaBySubjectVersion(r.Context(), subject, version)
 	if err != nil {
-		if errors.Is(err, storage.ErrSubjectNotFound) {
-			writeError(w, http.StatusNotFound, types.ErrorCodeSubjectNotFound, "Subject not found")
-			return
+		// If deleted=true and version not found, try to find the deleted version
+		if includeDeleted && (errors.Is(err, storage.ErrVersionNotFound) || errors.Is(err, storage.ErrSubjectNotFound)) {
+			schema, err = h.findDeletedVersion(r.Context(), subject, version)
 		}
-		if errors.Is(err, storage.ErrVersionNotFound) {
-			writeError(w, http.StatusNotFound, types.ErrorCodeVersionNotFound, "Version not found")
+		if err != nil {
+			if errors.Is(err, storage.ErrSubjectNotFound) {
+				writeError(w, http.StatusNotFound, types.ErrorCodeSubjectNotFound, "Subject not found")
+				return
+			}
+			if errors.Is(err, storage.ErrVersionNotFound) {
+				writeError(w, http.StatusNotFound, types.ErrorCodeVersionNotFound, "Version not found")
+				return
+			}
+			writeError(w, http.StatusInternalServerError, types.ErrorCodeInternalServerError, err.Error())
 			return
 		}
-		writeError(w, http.StatusInternalServerError, types.ErrorCodeInternalServerError, err.Error())
-		return
 	}
 
 	schemaStr := schema.Schema
@@ -282,6 +293,23 @@ func (h *Handler) GetVersion(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, http.StatusOK, resp)
 }
 
+// findDeletedVersion looks up a soft-deleted version by iterating all versions including deleted.
+func (h *Handler) findDeletedVersion(ctx context.Context, subject string, version int) (*storage.SchemaRecord, error) {
+	schemas, err := h.registry.GetSchemasBySubject(ctx, subject, true) // include deleted
+	if err != nil {
+		return nil, err
+	}
+	if len(schemas) == 0 {
+		return nil, storage.ErrSubjectNotFound
+	}
+	for _, s := range schemas {
+		if s.Version == version {
+			return s, nil
+		}
+	}
+	return nil, storage.ErrVersionNotFound
+}
+
 // RegisterSchema handles POST /subjects/{subject}/versions
 func (h *Handler) RegisterSchema(w http.ResponseWriter, r *http.Request) {
 	subject := chi.URLParam(r, "subject")
@@ -500,7 +528,7 @@ func (h *Handler) GetConfig(w http.ResponseWriter, r *http.Request) {
 		level, err := h.registry.GetSubjectConfig(r.Context(), subject)
 		if err != nil {
 			if errors.Is(err, storage.ErrNotFound) {
-				writeError(w, http.StatusNotFound, types.ErrorCodeSubjectNotFound,
+				writeError(w, http.StatusNotFound, types.ErrorCodeSubjectCompatNotFound,
 					fmt.Sprintf("Subject '%s' does not have subject-level compatibility configured", subject))
 				return
 			}
@@ -698,7 +726,7 @@ func (h *Handler) GetMode(w http.ResponseWriter, r *http.Request) {
 		mode, err := h.registry.GetSubjectMode(r.Context(), subject)
 		if err != nil {
 			if errors.Is(err, storage.ErrNotFound) {
-				writeError(w, http.StatusNotFound, types.ErrorCodeSubjectNotFound,
+				writeError(w, http.StatusNotFound, types.ErrorCodeSubjectModeNotFound,
 					fmt.Sprintf("Subject '%s' does not have subject-level mode configured", subject))
 				return
 			}
@@ -753,6 +781,32 @@ func (h *Handler) SetMode(w http.ResponseWriter, r *http.Request) {
 
 // parseVersion parses a version string, handling "latest" and "-1".
 // Returns errInvalidVersion for non-numeric strings, zero, or negative values (other than -1).
+// parsePagination extracts offset and limit query params and applies them to a slice length.
+// Returns the start and end indices for slicing.
+func parsePagination(r *http.Request, total int) (start, end int) {
+	start = 0
+	end = total
+
+	if offsetStr := r.URL.Query().Get("offset"); offsetStr != "" {
+		if o, err := strconv.Atoi(offsetStr); err == nil && o >= 0 {
+			start = o
+		}
+	}
+	if limitStr := r.URL.Query().Get("limit"); limitStr != "" {
+		if l, err := strconv.Atoi(limitStr); err == nil && l >= 0 {
+			end = start + l
+		}
+	}
+
+	if start > total {
+		start = total
+	}
+	if end > total {
+		end = total
+	}
+	return start, end
+}
+
 func parseVersion(s string) (int, error) {
 	if s == "latest" || s == "-1" {
 		return -1, nil
@@ -850,7 +904,9 @@ func (h *Handler) GetSubjectsBySchemaID(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
-	writeJSON(w, http.StatusOK, subjects)
+	// Apply pagination
+	start, end := parsePagination(r, len(subjects))
+	writeJSON(w, http.StatusOK, subjects[start:end])
 }
 
 // GetVersionsBySchemaID handles GET /schemas/ids/{id}/versions
@@ -887,7 +943,9 @@ func (h *Handler) GetVersionsBySchemaID(w http.ResponseWriter, r *http.Request)
 		})
 	}
 
-	writeJSON(w, http.StatusOK, result)
+	// Apply pagination
+	start, end := parsePagination(r, len(result))
+	writeJSON(w, http.StatusOK, result[start:end])
 }
 
 // ListSchemas handles GET /schemas

internal/api/types/types.go

@@ -136,6 +136,8 @@ const (
 	ErrorCodeSubjectNotSoftDeleted     = 40405
 	ErrorCodeSchemaVersionSoftDeleted  = 40406
 	ErrorCodeVersionNotSoftDeleted     = 40407
+	ErrorCodeSubjectCompatNotFound     = 40408
+	ErrorCodeSubjectModeNotFound       = 40409
 	ErrorCodeIncompatibleSchema        = 409
 	ErrorCodeInvalidSchema             = 42201
 	ErrorCodeInvalidSchemaType         = 42202

internal/compatibility/protobuf/checker.go

@@ -171,6 +171,11 @@ func (c *Checker) checkMessageCompatibility(newMsg, oldMsg protoreflect.MessageD
 		oldFields[int32(f.Number())] = f
 	}
 
+	// Track fields that move from non-oneof into a real oneof, keyed by oneof name.
+	// If multiple previously-independent fields land in the same oneof, that adds
+	// a mutual exclusion constraint which is an incompatible change.
+	fieldsMovedToOneof := make(map[string][]string) // oneof name -> list of field names
+
 	// Check each new field
 	for i := 0; i < newMsg.Fields().Len(); i++ {
 		newField := newMsg.Fields().Get(i)
@@ -187,16 +192,74 @@ func (c *Checker) checkMessageCompatibility(newMsg, oldMsg protoreflect.MessageD
 			continue
 		}
 
+		// Track fields moving into a real oneof from non-oneof
+		oldOneof := oldField.ContainingOneof()
+		newOneof := newField.ContainingOneof()
+		oldIsRealOneof := oldOneof != nil && !oldOneof.IsSynthetic()
+		newIsRealOneof := newOneof != nil && !newOneof.IsSynthetic()
+		if !oldIsRealOneof && newIsRealOneof {
+			oneofName := string(newOneof.Name())
+			fieldsMovedToOneof[oneofName] = append(fieldsMovedToOneof[oneofName], string(newField.Name()))
+		}
+
 		// Check field compatibility
 		c.checkFieldCompatibility(newField, oldField, msgName, result)
 		delete(oldFields, num)
 	}
 
-	// In proto3, field removal is wire-safe: readers ignore unknown fields and use
-	// defaults for missing ones. However, removing a field from a oneof IS
-	// incompatible because it changes oneof semantics.
+	// Check if fields were moved into an existing oneof.
+	// Moving a field into a oneof that already has other pre-existing members is
+	// incompatible (FIELD_MOVED_TO_EXISTING_ONEOF) because it adds a mutual
+	// exclusion constraint. Also, moving multiple previously-independent fields
+	// into the same new oneof creates a new mutual exclusion constraint.
+	//
+	// Build a set of old field numbers for quick lookup.
+	oldFieldNums := make(map[int32]bool)
+	for i := 0; i < oldMsg.Fields().Len(); i++ {
+		oldFieldNums[int32(oldMsg.Fields().Get(i).Number())] = true
+	}
+
+	for oneofName, movedFields := range fieldsMovedToOneof {
+		if len(movedFields) > 1 {
+			// Multiple independent fields moved into same oneof
+			result.AddMessage("Message '%s': multiple fields moved into oneof '%s', creating mutual exclusion",
+				msgName, oneofName)
+			continue
+		}
+		// Check if the target oneof has other members that already existed in
+		// the old schema (i.e. they're not brand new fields).
+		movedFieldName := movedFields[0]
+		otherPreExistingMember := false
+		for i := 0; i < newMsg.Fields().Len(); i++ {
+			nf := newMsg.Fields().Get(i)
+			no := nf.ContainingOneof()
+			if no == nil || no.IsSynthetic() || string(no.Name()) != oneofName {
+				continue
+			}
+			if string(nf.Name()) == movedFieldName {
+				continue
+			}
+			// Check if this field existed in the old schema by number
+			if oldFieldNums[int32(nf.Number())] {
+				otherPreExistingMember = true
+				break
+			}
+		}
+		if otherPreExistingMember {
+			result.AddMessage("Message '%s': field '%s' moved into existing oneof '%s'",
+				msgName, movedFieldName, oneofName)
+		}
+	}
+
+	// Check removed fields for compatibility issues:
+	// 1. Removing a required field (proto2) is always incompatible
+	// 2. Removing a field from a non-synthetic oneof is incompatible (changes oneof semantics)
+	// 3. In proto3, non-oneof field removal is wire-safe (readers ignore unknown fields)
 	for num, oldField := range oldFields {
-		if oldField.ContainingOneof() != nil {
+		if oldField.Cardinality() == protoreflect.Required {
+			result.AddMessage("Message '%s': required field '%s' (number %d) was removed",
+				msgName, oldField.Name(), num)
+		} else if oldField.ContainingOneof() != nil && !oldField.ContainingOneof().IsSynthetic() {
 			result.AddMessage("Message '%s': field '%s' (number %d) was removed from oneof",
 				msgName, oldField.Name(), num)
 		}
@@ -233,7 +296,14 @@ func (c *Checker) checkFieldCompatibility(newField, oldField protoreflect.FieldD
 	if oldCard != newCard {
 		// Some cardinality changes are compatible
 		if oldCard == protoreflect.Optional && newCard == protoreflect.Repeated {
-			// Optional to repeated - compatible for reading
+			// Per protobuf spec: "For string, bytes, and message fields, optional is
+			// compatible with repeated." These use length-delimited encoding which
+			// is the same wire format for both singular and repeated.
+			kind := oldField.Kind()
+			if kind != protoreflect.StringKind && kind != protoreflect.BytesKind && kind != protoreflect.MessageKind {
+				result.AddMessage("Message '%s': field '%s' changed from optional to repeated",
+					msgName, fieldName)
+			}
 		} else if oldCard == protoreflect.Required && newCard != protoreflect.Required {
 			// Required to optional/repeated - compatible
 		} else if newCard == protoreflect.Required && oldCard != protoreflect.Required {
@@ -255,9 +325,20 @@ func (c *Checker) checkFieldCompatibility(newField, oldField protoreflect.FieldD
 	// Check oneof membership changes
 	oldOneof := oldField.ContainingOneof()
 	newOneof := newField.ContainingOneof()
-	if (oldOneof == nil) != (newOneof == nil) {
-		result.AddMessage("Message '%s': field '%s' oneof membership changed",
-			msgName, fieldName)
+
+	// Determine if the oneofs are "real" (non-synthetic). Proto3 optional fields
+	// have a synthetic oneof wrapper that is invisible at the wire level.
+	oldIsRealOneof := oldOneof != nil && !oldOneof.IsSynthetic()
+	newIsRealOneof := newOneof != nil && !newOneof.IsSynthetic()
+
+	if oldIsRealOneof != newIsRealOneof {
+		if oldIsRealOneof && !newIsRealOneof {
+			// Moving OUT of a real oneof — incompatible (changes oneof semantics)
+			result.AddMessage("Message '%s': field '%s' oneof membership changed",
+				msgName, fieldName)
+		}
+		// Moving INTO a real oneof from non-oneof or synthetic oneof is compatible
+		// because the field number and wire format are preserved.
 	}
 }