Fail closed on malformed Business ID counter tables (#172)

* Validate Business ID counter table shape before allocation

* Cover short Business ID scope keys and nullable versions

* Preserve Business ID shape diagnostic after creation race

* Exercise malformed Business ID tables on live SQL providers

---------

Co-authored-by: Philip Z <hello@teaql.io>

Author: philipgreat

.github/workflows/live-dialects.yml

@@ -50,7 +50,7 @@ jobs:
       - name: Verify both live dialects (skips are forbidden)
         run: >-
           mvn -B -ntp -pl teaql-postgres,teaql-mysql,examples/business-id-runtime -am
-          -Dtest=PostgresIntegrationTest,MysqlIntegrationTest,JdbcBusinessIdLiveDialectIT#postgresAllocatesAcrossInstancesAndRestart+mysqlAllocatesAcrossInstancesAndRestart
+          -Dtest=PostgresIntegrationTest,MysqlIntegrationTest,JdbcBusinessIdLiveDialectIT#postgresAllocatesAcrossInstancesAndRestart+mysqlAllocatesAcrossInstancesAndRestart+postgresRejectsUnprotectedBusinessIdTable+mysqlRejectsUnprotectedBusinessIdTable
           -Dsurefire.failIfNoSpecifiedTests=false test
 
   business-id-sqlserver:
@@ -87,7 +87,7 @@ jobs:
       - name: Verify SQL Server Business ID schema and allocation (skips are forbidden)
         run: >-
           mvn -B -ntp -pl examples/business-id-runtime -am
-          '-Dtest=JdbcBusinessIdLiveDialectIT#sqlServerAllocatesAcrossInstancesAndRestart+sqlServerConcurrentSchemaStartupIsIdempotent'
+          '-Dtest=JdbcBusinessIdLiveDialectIT#sqlServerAllocatesAcrossInstancesAndRestart+sqlServerConcurrentSchemaStartupIsIdempotent+sqlServerRejectsUnprotectedBusinessIdTable'
           -Dsurefire.failIfNoSpecifiedTests=false test
 
   sqlserver:

examples/business-id-runtime/README.md

@@ -14,6 +14,17 @@ the context's `BusinessClock`; tests can replace that clock. The JDBC allocator
 uses root/aggregate/namespace/date scoping, optimistic allocation, bounded
 retry, and a range error when the configured digits are exhausted.
 
+For an existing counter table, `context.ensureSchema()` verifies the four
+required columns, a single-column `scope_key` primary or unique key, a text
+`scope_key` with capacity at least 512, and non-null 64-bit integral counter,
+version, and timestamp columns. It does not alter an incompatible table. A
+custom `TeaQLDatabase` adapter must return JDBC `DATA_TYPE`, `TYPE_NAME`,
+`COLUMN_SIZE`, `DECIMAL_DIGITS`, and `NULLABLE` metadata under the lower-case
+keys `data_type`, `type_name`, `column_size`, `decimal_digits`, and `nullable`
+in `getTableColumns`, and implement `getTablePrimaryKeyColumns` or
+`getTableUniqueKeys`. Missing shape metadata fails closed rather than letting
+allocation fail later.
+
 `BusinessIdGenerator`, `UserContext.generateBusinessId(...)`,
 `InMemoryBusinessIdGenerator`, and `JdbcBusinessIdGenerator` are deprecated
 compatibility APIs. They retain the old `PREFIXyyyyMMddNNNN` format and separate

examples/business-id-runtime/src/test/java/io/teaql/examples/businessid/BusinessIdRuntimeExampleTest.java

@@ -259,6 +259,96 @@ public void sqliteAllocatorRejectsIncompleteExistingSchema() throws Exception {
         }
     }
 
+    @Test
+    public void sqliteAllocatorRejectsExistingTableWithoutUniqueScopeKey() throws Exception {
+        assertInvalidExistingTable(
+                "CREATE TABLE teaql_business_id_space ("
+                        + "scope_key VARCHAR(512), current_value BIGINT NOT NULL, "
+                        + "version BIGINT NOT NULL, updated_at BIGINT NOT NULL)",
+                "scope_key");
+    }
+
+    @Test
+    public void sqliteAllocatorRejectsNullableCounter() throws Exception {
+        assertInvalidExistingTable(
+                "CREATE TABLE teaql_business_id_space ("
+                        + "scope_key VARCHAR(512) PRIMARY KEY, current_value BIGINT, "
+                        + "version BIGINT NOT NULL, updated_at BIGINT NOT NULL)",
+                "current_value");
+    }
+
+    @Test
+    public void sqliteAllocatorRejectsNarrowCounter() throws Exception {
+        assertInvalidExistingTable(
+                "CREATE TABLE teaql_business_id_space ("
+                        + "scope_key VARCHAR(512) PRIMARY KEY, current_value SMALLINT NOT NULL, "
+                        + "version BIGINT NOT NULL, updated_at BIGINT NOT NULL)",
+                "current_value");
+    }
+
+    @Test
+    public void sqliteAllocatorRejectsShortScopeKeyColumn() throws Exception {
+        assertInvalidExistingTable(
+                "CREATE TABLE teaql_business_id_space ("
+                        + "scope_key VARCHAR(64) PRIMARY KEY, current_value BIGINT NOT NULL, "
+                        + "version BIGINT NOT NULL, updated_at BIGINT NOT NULL)",
+                "scope_key");
+    }
+
+    @Test
+    public void sqliteAllocatorRejectsNullableVersion() throws Exception {
+        assertInvalidExistingTable(
+                "CREATE TABLE teaql_business_id_space ("
+                        + "scope_key VARCHAR(512) PRIMARY KEY, current_value BIGINT NOT NULL, "
+                        + "version BIGINT, updated_at BIGINT NOT NULL)",
+                "version");
+    }
+
+    @Test
+    public void sqliteAllocatorAcceptsNonNullableUniqueScopeKey() throws Exception {
+        Path databasePath = Files.createTempFile("teaql-business-id-unique-scope", ".db");
+        try (Connection connection = open(databasePath)) {
+            try (Statement statement = connection.createStatement()) {
+                statement.execute("CREATE TABLE teaql_business_id_space ("
+                        + "scope_key VARCHAR(512) NOT NULL UNIQUE, "
+                        + "current_value BIGINT NOT NULL, version BIGINT NOT NULL, "
+                        + "updated_at BIGINT NOT NULL)");
+            }
+            JdbcBusinessIdAllocator allocator = new JdbcBusinessIdAllocator(database(connection));
+            UserContext context = context(LocalDate.of(2026, 9, 20), allocator);
+            context.putAttribute(SchemaExecutor.class.getName(), noOpSchemaExecutor());
+            context.ensureSchema();
+            context.ensureSchema();
+            BusinessIdPlan plan = new DailySequenceBusinessIdProfile().plan(
+                    new BusinessIdGenerationRequest(ORDER_NUMBER, "tenant-a",
+                            "commerce_order", LocalDate.of(2026, 9, 20)));
+            Assert.assertEquals(1L, allocator.allocate(plan).sequence());
+            Assert.assertEquals(2L, allocator.allocate(plan).sequence());
+        } finally {
+            Files.deleteIfExists(databasePath);
+        }
+    }
+
+    private static void assertInvalidExistingTable(String ddl, String expectedField)
+            throws Exception {
+        Path databasePath = Files.createTempFile("teaql-business-id-invalid-shape", ".db");
+        try (Connection connection = open(databasePath)) {
+            try (Statement statement = connection.createStatement()) {
+                statement.execute(ddl);
+            }
+            JdbcBusinessIdAllocator allocator = new JdbcBusinessIdAllocator(database(connection));
+            UserContext context = context(LocalDate.of(2026, 9, 20), allocator);
+            context.putAttribute(SchemaExecutor.class.getName(), noOpSchemaExecutor());
+
+            IllegalStateException failure = Assert.assertThrows(
+                    IllegalStateException.class, context::ensureSchema);
+            Assert.assertTrue(failure.getMessage(),
+                    failure.getMessage().contains(expectedField));
+        } finally {
+            Files.deleteIfExists(databasePath);
+        }
+    }
+
     @Test
     public void sqliteAllocatorRejectsExhaustedAndOverflowedSequencesWithoutMutation()
             throws Exception {
@@ -447,7 +537,14 @@ public List<Map<String, Object>> getTableColumns(String tableName) {
                         try (ResultSet result = metadata.getColumns(
                                 connection.getCatalog(), connection.getSchema(), name, null)) {
                             while (result.next()) {
-                                columns.add(Map.of("column_name", result.getString("COLUMN_NAME")));
+                                Map<String, Object> column = new HashMap<>();
+                                column.put("column_name", result.getString("COLUMN_NAME"));
+                                column.put("data_type", result.getInt("DATA_TYPE"));
+                                column.put("type_name", result.getString("TYPE_NAME"));
+                                column.put("column_size", result.getInt("COLUMN_SIZE"));
+                                column.put("decimal_digits", result.getInt("DECIMAL_DIGITS"));
+                                column.put("nullable", result.getInt("NULLABLE"));
+                                columns.add(column);
                             }
                         }
                         if (!columns.isEmpty()) break;
@@ -458,6 +555,55 @@ public List<Map<String, Object>> getTableColumns(String tableName) {
                 }
             }
 
+            @Override
+            public Optional<List<String>> getTablePrimaryKeyColumns(String tableName) {
+                try {
+                    java.sql.DatabaseMetaData metadata = connection.getMetaData();
+                    for (String name : List.of(tableName, tableName.toUpperCase(Locale.ROOT))) {
+                        TreeMap<Short, String> ordered = new TreeMap<>();
+                        try (ResultSet result = metadata.getPrimaryKeys(
+                                connection.getCatalog(), connection.getSchema(), name)) {
+                            while (result.next()) {
+                                ordered.put(result.getShort("KEY_SEQ"),
+                                        result.getString("COLUMN_NAME"));
+                            }
+                        }
+                        if (!ordered.isEmpty()) return Optional.of(List.copyOf(ordered.values()));
+                    }
+                    return Optional.of(List.of());
+                } catch (SQLException error) {
+                    throw new RuntimeException(error);
+                }
+            }
+
+            @Override
+            public Optional<List<List<String>>> getTableUniqueKeys(String tableName) {
+                try {
+                    java.sql.DatabaseMetaData metadata = connection.getMetaData();
+                    for (String name : List.of(tableName, tableName.toUpperCase(Locale.ROOT))) {
+                        Map<String, TreeMap<Short, String>> indexes = new HashMap<>();
+                        try (ResultSet result = metadata.getIndexInfo(
+                                connection.getCatalog(), connection.getSchema(), name,
+                                true, true)) {
+                            while (result.next()) {
+                                String index = result.getString("INDEX_NAME");
+                                String column = result.getString("COLUMN_NAME");
+                                if (index == null || column == null) continue;
+                                indexes.computeIfAbsent(index, ignored -> new TreeMap<>())
+                                        .put(result.getShort("ORDINAL_POSITION"), column);
+                            }
+                        }
+                        if (!indexes.isEmpty()) {
+                            return Optional.of(indexes.values().stream()
+                                    .map(ordered -> List.copyOf(ordered.values())).toList());
+                        }
+                    }
+                    return Optional.of(List.of());
+                } catch (SQLException error) {
+                    throw new RuntimeException(error);
+                }
+            }
+
             private void bind(PreparedStatement statement, Object[] args) throws SQLException {
                 if (args == null) return;
                 for (int index = 0; index < args.length; index++) {

examples/business-id-runtime/src/test/java/io/teaql/examples/businessid/JdbcBusinessIdLiveDialectIT.java

@@ -10,6 +10,7 @@
 import io.teaql.core.sql.dialect.OracleDialect;
 import java.sql.Connection;
 import java.sql.DriverManager;
+import java.sql.Statement;
 import java.time.LocalDate;
 import java.util.ArrayList;
 import java.util.HashSet;
@@ -45,6 +46,21 @@ public void sqlServerAllocatesAcrossInstancesAndRestart() throws Exception {
         verify("MSSQL");
     }
 
+    @Test
+    public void postgresRejectsUnprotectedBusinessIdTable() throws Exception {
+        rejectUnprotectedTable("POSTGRES");
+    }
+
+    @Test
+    public void mysqlRejectsUnprotectedBusinessIdTable() throws Exception {
+        rejectUnprotectedTable("MYSQL");
+    }
+
+    @Test
+    public void sqlServerRejectsUnprotectedBusinessIdTable() throws Exception {
+        rejectUnprotectedTable("MSSQL");
+    }
+
     @Test
     public void dm8AllocatesAcrossInstancesAndRestart() throws Exception {
         verify("DM8");
@@ -176,6 +192,46 @@ private void verify(String dialect) throws Exception {
         }
     }
 
+    private void rejectUnprotectedTable(String dialect) throws Exception {
+        String prefix = "TEAQL_TEST_" + dialect + "_";
+        String url = System.getenv(prefix + "URL");
+        String user = System.getenv(prefix + "USER");
+        String password = System.getenv(prefix + "PASSWORD");
+        if (url == null || user == null || password == null) {
+            if (Boolean.parseBoolean(System.getenv("TEAQL_REQUIRE_LIVE_DB"))) {
+                Assert.fail(dialect + " Business ID live gate requires URL, USER, and PASSWORD");
+            }
+            Assume.assumeTrue(dialect + " live database is not configured", false);
+        }
+        Assert.assertTrue("Use a dedicated teaql_live_* database for " + dialect,
+                url.contains("teaql_live_"));
+
+        String table = "bid_shape_" + UUID.randomUUID().toString().substring(0, 8);
+        try (Connection connection = DriverManager.getConnection(url, user, password)) {
+            try (Statement statement = connection.createStatement()) {
+                statement.execute("CREATE TABLE " + table + " ("
+                        + "scope_key VARCHAR(512), current_value BIGINT NOT NULL, "
+                        + "version BIGINT NOT NULL, updated_at BIGINT NOT NULL)");
+            }
+            try {
+                JdbcBusinessIdAllocator allocator = new JdbcBusinessIdAllocator(
+                        BusinessIdRuntimeExampleTest.database(connection), table);
+                UserContext context = BusinessIdRuntimeExampleTest.context(
+                        BUSINESS_DATE, allocator);
+                context.putAttribute(SchemaExecutor.class.getName(),
+                        BusinessIdRuntimeExampleTest.noOpSchemaExecutor());
+                IllegalStateException failure = Assert.assertThrows(
+                        IllegalStateException.class, context::ensureSchema);
+                Assert.assertTrue(failure.getMessage(),
+                        failure.getMessage().contains("scope_key"));
+            } finally {
+                try (Statement statement = connection.createStatement()) {
+                    statement.execute("DROP TABLE " + table);
+                }
+            }
+        }
+    }
+
     private static JdbcBusinessIdAllocator allocator(Connection connection, String dialect) {
         if ("ORACLE".equals(dialect)) {
             return new JdbcBusinessIdAllocator(