diff --git a/.rubocop_cc.yml b/.rubocop_cc.yml
index 9a5b91f82ad..e23fc419548 100644
--- a/.rubocop_cc.yml
+++ b/.rubocop_cc.yml
@@ -126,12 +126,8 @@ Style/ExpandPathArguments:
   Exclude:
     - 'db/migrations/20130911111938_encrypt_app_env_json.rb'
     - 'db/migrations/201805*'
-Rails/DangerousColumnNames: # Disabled for old migrations as we cannot change content of those
-  Enabled: true
-  Exclude:
-    - !ruby/regexp /db/migrations/201([0-9]).+\.rb$/
-    - !ruby/regexp /db/migrations/202([0-2]).+\.rb$/
-    - db/migrations/20221125134500_add_request_count_table.rb
+Rails/DangerousColumnNames: # Disabled, in comparison to active_record we need to add the id column manually in sequel
+  Enabled: false
 Rails/SkipsModelValidations: # We don`t want any model at all in migrations and migration specs
   Enabled: true
   Exclude:
diff --git a/app/actions/labels_update.rb b/app/actions/labels_update.rb
index 9ff3d71a167..5f7adab786a 100644
--- a/app/actions/labels_update.rb
+++ b/app/actions/labels_update.rb
@@ -12,16 +12,19 @@ def update(resource, labels, label_klass, destroy_nil: true)
           prefix, name = VCAP::CloudController::MetadataHelpers.extract_prefix(label_key)
 
           if label_value.nil? && destroy_nil # Delete Label
-            label_klass.where(resource_guid: resource.guid, key_name: name).where(Sequel.or([[:key_prefix, prefix], [:key_prefix, prefix.to_s]])).try(:destroy)
+            label_klass.find(key_prefix: prefix.to_s, resource_guid: resource.guid, key_name: name)&.destroy
             next
           end
 
           begin
             tries ||= 2
             label_klass.db.transaction(savepoint: true) do
-              label = label_klass.where(resource_guid: resource.guid, key_name: name).where(Sequel.or([[:key_prefix, prefix], [:key_prefix, prefix.to_s]])).first
-              label ||= label_klass.create(resource_guid: resource.guid, key_name: name, key_prefix: prefix)
-              label.update(value: label_value)
+              label = label_klass.find(key_prefix: prefix.to_s, resource_guid: resource.guid, key_name: name)
+              if label.nil?
+                label_klass.create(resource_guid: resource.guid, key_name: name, key_prefix: prefix, value: label_value)
+              else
+                label.update(value: label_value)
+              end
             end
           rescue Sequel::UniqueConstraintViolation => e
             if (tries -= 1).positive?
diff --git a/app/fetchers/label_selector_query_generator.rb b/app/fetchers/label_selector_query_generator.rb
index 3f2440280e1..fde07a3e2bb 100644
--- a/app/fetchers/label_selector_query_generator.rb
+++ b/app/fetchers/label_selector_query_generator.rb
@@ -23,15 +23,13 @@ def add_selector_queries(label_klass:, resource_dataset:, requirements:, resourc
       def guids_for_set_inclusion(label_klass, requirement)
         label_klass.
           select(:resource_guid).
-          where(key_name: requirement.key_name, value: requirement.values).
-          where(Sequel.or([[:key_prefix, requirement.key_prefix], [:key_prefix, requirement.key_prefix.to_s]]))
+          where(key_prefix: requirement.key_prefix.to_s, key_name: requirement.key_name, value: requirement.values)
       end
 
       def guids_for_existence(label_klass, requirement)
         label_klass.
           select(:resource_guid).
-          where(key_name: requirement.key_name).
-          where(Sequel.or([[:key_prefix, requirement.key_prefix], [:key_prefix, requirement.key_prefix.to_s]]))
+          where(key_prefix: requirement.key_prefix.to_s, key_name: requirement.key_name)
       end
     end
   end
diff --git a/app/models/helpers/metadata_model_mixin.rb b/app/models/helpers/metadata_model_mixin.rb
index b906490c11a..f9245972645 100644
--- a/app/models/helpers/metadata_model_mixin.rb
+++ b/app/models/helpers/metadata_model_mixin.rb
@@ -4,12 +4,12 @@ def self.included(included_class)
       included_class.class_eval <<-RUBY, __FILE__, __LINE__ + 1
         # Transparently convert datatypes of key_prefix so empty strings are persisted in the DB instead of NULL
         def key_prefix
-           self[:key_prefix].presence
-         end
-         def key_prefix=(value)
-           self[:key_prefix] = value.nil? ? '' : value
-         end
-         def key_name=(val)
+          self[:key_prefix].presence
+        end
+        def key_prefix=(value)
+          self[:key_prefix] = value.nil? ? '' : value
+        end
+        def key_name=(val)
           self[:key_name] = val.nil? ? '' : val
         end
         def key_name
diff --git a/db/migrations/20231221123000_rename_annotations_key_column.rb b/db/migrations/20231221123000_rename_annotations_key_column.rb
index d45e86d8a0b..3bc743e9c57 100644
--- a/db/migrations/20231221123000_rename_annotations_key_column.rb
+++ b/db/migrations/20231221123000_rename_annotations_key_column.rb
@@ -1,32 +1,32 @@
-TABLE_BASE_NAMES = %w[
-  app
-  build
-  buildpack
-  deployment
-  domain
-  droplet
-  isolation_segment
-  organization
-  package
-  process
-  revision
-  route_binding
-  route
-  service_binding
-  service_broker
-  service_broker_update_request
-  service_instance
-  service_key
-  service_offering
-  service_plan
-  space
-  stack
-  task
-  user
-].freeze
-annotation_tables = TABLE_BASE_NAMES.map { |tbn| "#{tbn}_annotations" }.freeze
-
 Sequel.migration do
+  table_base_names = %w[
+    app
+    build
+    buildpack
+    deployment
+    domain
+    droplet
+    isolation_segment
+    organization
+    package
+    process
+    revision
+    route_binding
+    route
+    service_binding
+    service_broker
+    service_broker_update_request
+    service_instance
+    service_key
+    service_offering
+    service_plan
+    space
+    stack
+    task
+    user
+  ].freeze
+  annotation_tables = table_base_names.map { |tbn| "#{tbn}_annotations" }.freeze
+
   no_transaction # Disable automatic transactions
 
   up do
diff --git a/db/migrations/20240102150000_add_annotation_label_uniqueness.rb b/db/migrations/20240102150000_add_annotation_label_uniqueness.rb
new file mode 100644
index 00000000000..1dc3fc25194
--- /dev/null
+++ b/db/migrations/20240102150000_add_annotation_label_uniqueness.rb
@@ -0,0 +1,126 @@
+require 'logger'
+
+def unique_index_name(table)
+  :"#{table}_unique"
+end
+
+def with_dropped_view(table)
+  # Recreate view just for postgres as it cannot alter types while a view is on the table
+  if database_type == :postgres
+    drop_view(:"#{table}_migration_view", if_exists: true)
+    yield if block_given?
+    create_view(:"#{table}_migration_view", self[table.to_sym].select do
+      [id, guid, created_at, updated_at, resource_guid, key_prefix, key_name, value]
+    end)
+  elsif block_given?
+    yield
+  end
+end
+
+Sequel.migration do
+  table_base_names = %w[
+    app
+    build
+    buildpack
+    deployment
+    domain
+    droplet
+    isolation_segment
+    organization
+    package
+    process
+    revision
+    route_binding
+    route
+    service_binding
+    service_broker
+    service_broker_update_request
+    service_instance
+    service_key
+    service_offering
+    service_plan
+    space
+    stack
+    task
+    user
+  ].freeze
+  annotation_tables = table_base_names.map { |tbn| "#{tbn}_annotations" }.freeze
+  label_tables = table_base_names.map { |tbn| "#{tbn}_labels" }.freeze
+
+  no_transaction # Disable automatic transactions
+
+  up do
+    (annotation_tables + label_tables).each do |table|
+      transaction do
+        # Create Temporary table for use later on
+        create_table! :"#{table}_temp", temp: true do
+          primary_key :id, name: :id
+          Integer :min_id, null: false
+        end
+
+        # Just allow selects on this table while the migration runs for full consistency
+        run "LOCK TABLE #{table}, #{table}_temp IN SHARE MODE;" if database_type == :postgres
+        run "LOCK TABLES #{table} WRITE, #{table}_temp WRITE;" if database_type == :mysql
+
+        # Updating the temporary column with truncated keys(should never chop of anything since the api just allows 63 chars)
+        # We run this in the DB as to minimize the time we hold the share mode lock on the table
+        self[table.to_sym].update(key_name: Sequel::SQL::Function.new(:SUBSTR, :key_name, 1, 63))
+
+        # Make en empty string the default for key_prefix as null in the unique constraint would not work.
+        # Null values are not equal to other Null values so a row that has NULL can be a duplicate then.
+        self[table.to_sym].where(key_prefix: nil).update(key_prefix: '')
+        self[table.to_sym].where(key_name: nil).update(key_name: '')
+
+        # Recreate view just for postgres as it cannot alter types while a view is on the table
+        with_dropped_view(table) do
+          alter_table(table.to_sym) do
+            set_column_default :key_prefix, ''
+            set_column_not_null :key_prefix
+            set_column_not_null :key_name
+            set_column_type :key_name, String, size: 63
+          end
+        end
+
+        # Delete duplicates (in the DB as doing it in ruby is slow), we need to use a temporary table
+        # as mysql doesnt allow subselects on the same table it deletes from
+        min_ids = from(table.to_sym).
+                  select(Sequel.function(:MIN, :id).as(:min_id)).
+                  group_by(:resource_guid, :key_prefix, :key_name)
+        self[:"#{table}_temp"].import([:min_id], min_ids)
+        self[table.to_sym].exclude(id: from(:"#{table}_temp").select(:min_id)).delete
+
+        # Add unique constraint if not already present
+        if indexes(table.to_sym)[unique_index_name(table)].nil?
+          alter_table(table.to_sym) do
+            add_unique_constraint %i[resource_guid key_prefix key_name], name: unique_index_name(table)
+          end
+        end
+      ensure
+        # Be sure to unlock the table on errors as this does not happen automatically by rolling back a transaction mysql
+        run 'UNLOCK TABLES;' if database_type == :mysql
+      end
+    end
+  end
+
+  down do
+    (annotation_tables + label_tables).each do |table|
+      transaction do
+        # Drop unique constraint
+        if indexes(table.to_sym)[unique_index_name(table)].present?
+          alter_table(table.to_sym) do
+            drop_constraint(unique_index_name(table), type: :unique)
+          end
+        end
+        # Revert default type in key_prefix and null values handling
+        with_dropped_view(table) do
+          alter_table(table.to_sym) do
+            set_column_allow_null :key_prefix
+            set_column_allow_null :key_name
+            set_column_default :key_prefix, nil
+            set_column_type :key_name, String, size: 1000 if table.end_with?('_annotations')
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/spec/migrations/20231221123000_rename_annotations_key_column_spec.rb b/spec/migrations/20231221123000_rename_annotations_key_column_spec.rb
index e9b7292a13a..46a7990bc37 100644
--- a/spec/migrations/20231221123000_rename_annotations_key_column_spec.rb
+++ b/spec/migrations/20231221123000_rename_annotations_key_column_spec.rb
@@ -1,10 +1,11 @@
 require 'spec_helper'
+require 'migrations/helpers/migration_shared_context'
+
+RSpec.describe 'migration to streamline changes to annotation_key_prefix', isolation: :truncation, type: :migration do
+  include_context 'migration' do
+    let(:migration_filename) { '20231221123000_rename_annotations_key_column.rb' }
+  end
 
-RSpec.describe 'migration to streamline changes to annotation_key_prefix', isolation: :truncation do
-  let(:filename) { '20231221123000_rename_annotations_key_column.rb' }
-  let(:tmp_down_migrations_dir) { Dir.mktmpdir }
-  let(:tmp_up_migrations_dir) { Dir.mktmpdir }
-  let(:db) { Sequel::Model.db }
   let(:tables) do
     %w[
       app
@@ -36,33 +37,13 @@
 
   let(:annotation_tables) { tables.map { |tbn| "#{tbn}_annotations" }.freeze }
 
-  before do
-    Sequel.extension :migration
-    # Find all migrations
-    migration_files = Dir.glob("#{DBMigrator::SEQUEL_MIGRATIONS}/*.rb")
-    # Calculate the index of our migration file we`d  like to test
-    migration_index = migration_files.find_index { |file| file.end_with?(filename) }
-    # Make a file list of the migration file we like to test plus all migrations after the one we want to test
-    migration_files_after_test = migration_files[migration_index...]
-    # Copy them to a temp directory
-    FileUtils.cp(migration_files_after_test, tmp_down_migrations_dir)
-    FileUtils.cp(File.join(DBMigrator::SEQUEL_MIGRATIONS, filename), tmp_up_migrations_dir)
-    # Revert the given migration and everything newer so we are at the database version exactly before our migration we want to test.
-    Sequel::Migrator.run(db, tmp_down_migrations_dir, target: 0, allow_missing_migration_files: true)
-  end
-
-  after do
-    FileUtils.rm_rf(tmp_up_migrations_dir)
-    FileUtils.rm_rf(tmp_down_migrations_dir)
-  end
-
   describe 'annotation tables' do
     it 'has renamed the column key to key_name' do
       annotation_tables.each do |table|
         expect(db[table.to_sym].columns).to include(:key)
         expect(db[table.to_sym].columns).not_to include(:key_name)
       end
-      expect { Sequel::Migrator.run(db, tmp_up_migrations_dir, allow_missing_migration_files: true) }.not_to raise_error
+      expect { Sequel::Migrator.run(db, migration_to_test, allow_missing_migration_files: true) }.not_to raise_error
       annotation_tables.each do |table|
         expect(db[table.to_sym].columns).not_to include(:key)
         expect(db[table.to_sym].columns).to include(:key_name)
diff --git a/spec/migrations/20240102150000_add_annotation_label_uniqueness_spec.rb b/spec/migrations/20240102150000_add_annotation_label_uniqueness_spec.rb
new file mode 100644
index 00000000000..9c38a976dae
--- /dev/null
+++ b/spec/migrations/20240102150000_add_annotation_label_uniqueness_spec.rb
@@ -0,0 +1,237 @@
+require 'spec_helper'
+require 'migrations/helpers/migration_shared_context'
+
+RSpec.describe 'migration to add unique constraint to annotation and labels', isolation: :truncation, type: :migration do
+  include_context 'migration' do
+    let(:migration_filename) { '20240102150000_add_annotation_label_uniqueness.rb' }
+  end
+
+  let(:isolation_segment) { VCAP::CloudController::IsolationSegmentModel }
+  let(:annotation) { VCAP::CloudController::IsolationSegmentAnnotationModel }
+  let(:label) { VCAP::CloudController::IsolationSegmentLabelModel }
+
+  describe 'annotation tables' do
+    it 'truncates keys to 63 characters' do
+      i1 = isolation_segment.create(name: 'bommel')
+      key_name = 'a' * 64
+      truncated_key_name = 'a' * 63
+
+      a1 = annotation.create(resource_guid: i1.guid, key_name: key_name, value: 'some_value')
+
+      expect { Sequel::Migrator.run(db, migration_to_test, allow_missing_migration_files: true) }.not_to raise_error
+      expect(a1.reload.key_name).to eq(truncated_key_name)
+    end
+
+    it 'leaves keys that are shorter than 64 characters unchanged' do
+      i1 = isolation_segment.create(name: 'bommel')
+      key_name = 'a' * 63
+
+      a1 = annotation.create(resource_guid: i1.guid, key_name: key_name, value: 'some_value')
+
+      expect { Sequel::Migrator.run(db, migration_to_test, allow_missing_migration_files: true) }.not_to raise_error
+      expect(a1.reload.key_name).to eq(key_name)
+    end
+
+    it 'removes duplicate annotations but keeps one with smallest id' do
+      i1 = isolation_segment.create(name: 'bommel')
+      key_name = 'a' * 63
+
+      # In case key_prefix is not set
+      a1 = annotation.create(resource_guid: i1.guid, key_name: key_name, value: 'v1')
+      a2 = annotation.create(resource_guid: i1.guid, key_name: key_name, value: 'v2')
+      a3 = annotation.create(resource_guid: i1.guid, key_name: key_name, value: 'v3')
+      # In case key_prefix is set
+      b1 = annotation.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key_name, value: 'v1')
+      b2 = annotation.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key_name, value: 'v2')
+      b3 = annotation.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key_name, value: 'v3')
+
+      expect(a1.id).to be < a2.id
+      expect(a1.id).to be < a3.id
+      expect(b1.id).to be < b2.id
+      expect(b1.id).to be < b3.id
+
+      expect { Sequel::Migrator.run(db, migration_to_test, allow_missing_migration_files: true) }.not_to raise_error
+      expect(annotation.where(key_name:).count).to eq(2)
+      expect(a1.reload).to be_a(annotation)
+      expect { a2.reload }.to raise_error(Sequel::NoExistingObject)
+      expect { a3.reload }.to raise_error(Sequel::NoExistingObject)
+      expect(b1.reload).to be_a(annotation)
+      expect { b2.reload }.to raise_error(Sequel::NoExistingObject)
+      expect { b3.reload }.to raise_error(Sequel::NoExistingObject)
+    end
+
+    it 'does not remove records if any column of key, key_prefix or resource_guid is different' do
+      i1 = isolation_segment.create(name: 'bommel')
+      i2 = isolation_segment.create(name: 'sword')
+      key_a = 'a' * 63
+      key_b = 'b' * 63
+
+      # In case key_prefix is not set
+      a1 = annotation.create(resource_guid: i1.guid, key_name: key_a, value: 'v1')
+      a2 = annotation.create(resource_guid: i2.guid, key_name: key_a, value: 'v2')
+      a3 = annotation.create(resource_guid: i1.guid, key_name: key_b, value: 'v3')
+      # In case key_prefix is set
+      b1 = annotation.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key_a, value: 'v1')
+      b2 = annotation.create(resource_guid: i2.guid, key_prefix: 'bommel', key_name: key_a, value: 'v2')
+      b3 = annotation.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key_b, value: 'v3')
+      b4 = annotation.create(resource_guid: i1.guid, key_prefix: 'sword', key_name: key_a, value: 'v4')
+
+      expect { Sequel::Migrator.run(db, migration_to_test, allow_missing_migration_files: true) }.not_to raise_error
+
+      expect(annotation.all.count).to eq(7)
+      expect(a1.reload).to be_a(annotation)
+      expect(a2.reload).to be_a(annotation)
+      expect(a3.reload).to be_a(annotation)
+      expect(b1.reload).to be_a(annotation)
+      expect(b2.reload).to be_a(annotation)
+      expect(b3.reload).to be_a(annotation)
+      expect(b4.reload).to be_a(annotation)
+    end
+
+    it 'does not allow adding a duplicate' do
+      i1 = isolation_segment.create(name: 'bommel')
+      i2 = isolation_segment.create(name: 'sword')
+      key = 'a' * 63
+
+      # In case key_prefix is not set
+      annotation.create(resource_guid: i1.guid, key_name: key, value: 'v1')
+      # In case key_prefix is set
+      annotation.create(resource_guid: i2.guid, key_prefix: 'bommel', key_name: key, value: 'v1')
+
+      expect { Sequel::Migrator.run(db, migration_to_test, allow_missing_migration_files: true) }.not_to raise_error
+
+      expect { annotation.create(resource_guid: i1.guid, key_name: key, value: 'v2') }.to raise_error(Sequel::UniqueConstraintViolation)
+      expect { annotation.create(resource_guid: i2.guid, key_prefix: 'bommel', key_name: key, value: 'v2') }.to raise_error(Sequel::UniqueConstraintViolation)
+    end
+
+    it 'does allow adding a different annotation' do
+      i1 = isolation_segment.create(name: 'bommel')
+      i2 = isolation_segment.create(name: 'sword')
+      key_a = 'a' * 63
+      key_b = 'b' * 63
+
+      expect { Sequel::Migrator.run(db, migration_to_test, allow_missing_migration_files: true) }.not_to raise_error
+
+      # In case key_prefix is not set
+      a1 = annotation.create(resource_guid: i1.guid, key_name: key_a, value: 'v1')
+      a2 = annotation.create(resource_guid: i2.guid, key_name: key_a, value: 'v2')
+      a3 = annotation.create(resource_guid: i1.guid, key_name: key_b, value: 'v3')
+      # In case key_prefix is set
+      b1 = annotation.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key_a, value: 'v1')
+      b2 = annotation.create(resource_guid: i2.guid, key_prefix: 'bommel', key_name: key_a, value: 'v2')
+      b3 = annotation.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key_b, value: 'v3')
+      b4 = annotation.create(resource_guid: i1.guid, key_prefix: 'sword', key_name: key_a, value: 'v4')
+
+      expect(annotation.all.count).to eq(7)
+      expect(a1.reload).to be_a(annotation)
+      expect(a2.reload).to be_a(annotation)
+      expect(a3.reload).to be_a(annotation)
+      expect(b1.reload).to be_a(annotation)
+      expect(b2.reload).to be_a(annotation)
+      expect(b3.reload).to be_a(annotation)
+      expect(b4.reload).to be_a(annotation)
+    end
+  end
+
+  describe 'labels tables' do
+    it 'removes duplicate annotations but keeps one with smallest id' do
+      i1 = isolation_segment.create(name: 'bommel')
+      key = 'a' * 63
+
+      # In case key_prefix is not set
+      a1 = label.create(resource_guid: i1.guid, key_name: key, value: 'v1')
+      a2 = label.create(resource_guid: i1.guid, key_name: key, value: 'v2')
+      a3 = label.create(resource_guid: i1.guid, key_name: key, value: 'v3')
+      # In case key_prefix is set
+      b1 = label.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key, value: 'v1')
+      b2 = label.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key, value: 'v2')
+      b3 = label.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key, value: 'v3')
+
+      expect(a1.id).to be < a2.id
+      expect(a1.id).to be < a3.id
+      expect(b1.id).to be < b2.id
+      expect(b1.id).to be < b3.id
+
+      expect { Sequel::Migrator.run(db, migration_to_test, allow_missing_migration_files: true) }.not_to raise_error
+      expect(label.where(key_name: key).count).to eq(2)
+      expect(a1.reload).to be_a(label)
+      expect { a2.reload }.to raise_error(Sequel::NoExistingObject)
+      expect { a3.reload }.to raise_error(Sequel::NoExistingObject)
+      expect(b1.reload).to be_a(label)
+      expect { b2.reload }.to raise_error(Sequel::NoExistingObject)
+      expect { b3.reload }.to raise_error(Sequel::NoExistingObject)
+    end
+
+    it 'does not remove records if any column of key_name, key_prefix or resource_guid is different' do
+      i1 = isolation_segment.create(name: 'bommel')
+      i2 = isolation_segment.create(name: 'sword')
+      key_a = 'a' * 63
+      key_b = 'b' * 63
+
+      # In case key_prefix is not set
+      a1 = label.create(resource_guid: i1.guid, key_name: key_a, value: 'v1')
+      a2 = label.create(resource_guid: i2.guid, key_name: key_a, value: 'v2')
+      a3 = label.create(resource_guid: i1.guid, key_name: key_b, value: 'v3')
+      # In case key_prefix is set
+      b1 = label.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key_a, value: 'v1')
+      b2 = label.create(resource_guid: i2.guid, key_prefix: 'bommel', key_name: key_a, value: 'v2')
+      b3 = label.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key_b, value: 'v3')
+      b4 = label.create(resource_guid: i1.guid, key_prefix: 'sword', key_name: key_a, value: 'v4')
+
+      expect { Sequel::Migrator.run(db, migration_to_test, allow_missing_migration_files: true) }.not_to raise_error
+
+      expect(label.all.count).to eq(7)
+      expect(a1.reload).to be_a(label)
+      expect(a2.reload).to be_a(label)
+      expect(a3.reload).to be_a(label)
+      expect(b1.reload).to be_a(label)
+      expect(b2.reload).to be_a(label)
+      expect(b3.reload).to be_a(label)
+      expect(b4.reload).to be_a(label)
+    end
+
+    it 'does not allow adding a duplicate' do
+      i1 = isolation_segment.create(name: 'bommel')
+      i2 = isolation_segment.create(name: 'sword')
+      key = 'a' * 63
+
+      # In case key_prefix is not set
+      label.create(resource_guid: i1.guid, key_name: key, value: 'v1')
+      # In case key_prefix is set
+      label.create(resource_guid: i2.guid, key_prefix: 'bommel', key_name: key, value: 'v1')
+
+      expect { Sequel::Migrator.run(db, migration_to_test, allow_missing_migration_files: true) }.not_to raise_error
+
+      expect { label.create(resource_guid: i1.guid, key_name: key, value: 'v2') }.to raise_error(Sequel::UniqueConstraintViolation)
+      expect { label.create(resource_guid: i2.guid, key_prefix: 'bommel', key_name: key, value: 'v2') }.to raise_error(Sequel::UniqueConstraintViolation)
+    end
+
+    it 'does allow adding a different label' do
+      i1 = isolation_segment.create(name: 'bommel')
+      i2 = isolation_segment.create(name: 'sword')
+      key_a = 'a' * 63
+      key_b = 'b' * 63
+
+      expect { Sequel::Migrator.run(db, migration_to_test, allow_missing_migration_files: true) }.not_to raise_error
+
+      # In case key_prefix is not set
+      a1 = label.create(resource_guid: i1.guid, key_name: key_a, value: 'v1')
+      a2 = label.create(resource_guid: i2.guid, key_name: key_a, value: 'v2')
+      a3 = label.create(resource_guid: i1.guid, key_name: key_b, value: 'v3')
+      # In case key_prefix is set
+      b1 = label.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key_a, value: 'v1')
+      b2 = label.create(resource_guid: i2.guid, key_prefix: 'bommel', key_name: key_a, value: 'v2')
+      b3 = label.create(resource_guid: i1.guid, key_prefix: 'bommel', key_name: key_b, value: 'v3')
+      b4 = label.create(resource_guid: i1.guid, key_prefix: 'sword', key_name: key_a, value: 'v4')
+
+      expect(label.all.count).to eq(7)
+      expect(a1.reload).to be_a(label)
+      expect(a2.reload).to be_a(label)
+      expect(a3.reload).to be_a(label)
+      expect(b1.reload).to be_a(label)
+      expect(b2.reload).to be_a(label)
+      expect(b3.reload).to be_a(label)
+      expect(b4.reload).to be_a(label)
+    end
+  end
+end