fix(cassandra): resolve concurrent idempotency race and remove dropped tables from cleanup

- Re-check findSchemaInSubject on CAS retry to detect concurrent
  registrations of the same schema (fixes TestSchemaIdempotency)
- Remove schemas_by_fingerprint and subjects from BDD truncation list
  (tables were dropped in SAI migration)

Author: millerjp

internal/storage/cassandra/store.go

@@ -477,6 +477,17 @@ func (s *Store) CreateSchema(ctx context.Context, record *storage.SchemaRecord)
 
 	// Allocate next subject version and persist atomically.
 	for attempt := 0; attempt < s.cfg.MaxRetries; attempt++ {
+		// Re-check on retry: a concurrent writer may have registered this schema
+		if attempt > 0 {
+			if existing, err := s.findSchemaInSubject(ctx, record.Subject, schemaID); err == nil && existing != nil {
+				record.ID = existing.ID
+				record.Version = existing.Version
+				record.Fingerprint = fp
+				record.CreatedAt = existing.CreatedAt
+				return storage.ErrSchemaExists
+			}
+		}
+
 		latestVersion, latestSchemaID, exists, err := s.getSubjectLatest(ctx, record.Subject)
 		if err != nil {
 			return err

tests/bdd/bdd_test.go

@@ -305,8 +305,8 @@ func cleanCassandra() error {
 		"users_by_email", "users_by_id",
 		"id_alloc", "modes", "global_config", "subject_configs",
 		"references_by_target", "schema_references",
-		"subjects", "subject_latest", "subject_versions",
-		"schemas_by_fingerprint", "schemas_by_id",
+		"subject_latest", "subject_versions",
+		"schemas_by_id",
 	}
 	for _, t := range tables {
 		if err := session.Query("TRUNCATE " + t).Exec(); err != nil {