Issue3609: Introduce Blacklist refactoring via exclusion on database level

factcast-site/documentation-docsy/content/en/About/migration.md

@@ -4,6 +4,31 @@ type = "docs"
 weight = 100015
 +++
 
+## Upgrading to 0.12.0
+
+### New exclusion mechanism replaces former "Blacklist"
+
+Maintaining a separate table of factIds to be filtered out of every result turned out to be inefficient.
+Therefore, a new `exclusion_reason` column is introduced on the fact table, causing all facts with a non-NULL
+value to be ignored at query time. To migrate from the previous solution, follow the steps described below:
+
+1. The feature is guarded behind the new property `factcast.store.useInternalExclusion`, which defaults to `false`.
+   Ensure FactCast is deployed once with this default to trigger the changeset, which migrates all entries from the blacklist into the new column and creates a new partial GIN index on the fact header that takes the new column into account.
+   1. For tables with more than 10,000,000 entries, the automated migration and index creation are skipped and have to be conducted manually:
+      1. ```sql
+         -- adjust the batch size depending on your needs
+         CALL migrate_blacklist_to_exclusion_reason(10000);
+         ```
+      2. ```sql
+         -- After the migration is done create the new index
+         CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_fact_header_active
+         ON fact USING GIN (header jsonb_path_ops)
+         WITH (fastupdate=false)
+         WHERE exclusion_reason IS NULL;
+         ```
+2. To switch to the new behavior, deploy FactCast again with `factcast.store.useInternalExclusion` set to `true`. Attempts to add new entries to the old blacklist will then trigger a warning.
+3. Finally, execute the `drop_idx_fact_header.sql` changeset to avoid maintaining duplicate indexes on each insert. Please be aware that this index has to be re-created before rolling back to a previous version.
+
 ## Upgrading to 0.11.0
 
 ### `Projection.postprocess` now takes a `Collection` instead of a `List`

factcast-site/documentation-docsy/content/en/Concept/blacklisting.md → factcast-site/documentation-docsy/content/en/Concept/excluding-facts.md

@@ -1,26 +1,50 @@
 ---
-title: "Blacklisting"
+title: "Excluding Facts"
 weight: 150
 type: docs
 ---
 
+> Prior to the migration to version X.X this mechanism was called `blacklisting`.
+
 In rare occasions it can happen that one or more facts were emitted that are broken in a way that makes it necessary to
 remove them from the fact stream altogether. Events including or referencing malware might be an example.
 
-**Blacklisting** provides a way to prevent single facts from being delivered to
+**Exclusion** provides a way to prevent single facts from being delivered to
 any consumers, without the need to actually delete them from the history.
 
 {{% alert title="A word of caution" color="warning" %}}
 Please remember that removing or altering facts that already got emitted from the fact stream, no matter if through
-deletion or blacklisting, should be avoided whenever it is possible as this contradicts the core principle in
-event-sourcing that facts are immutable. Also remember that just blacklisting a fact won't revert that consumers might
-have processed and reacted to that fact already and removing it later might prevent reproducing the current state of the
-system.
+deletion or exclusion, should be avoided whenever it is possible as this contradicts the core principle in
+event-sourcing that facts are immutable. Also remember that excluding a fact won't revert that consumers might
+have processed and reacted to that fact already. Therefor, removing via exclusion might prevent reproducing the current
+state of the system.
 {{% /alert %}}
 
-If nevertheless you need to blacklist facts, there are two options:
+If nevertheless you need to exclude facts, you can do so as follows:
+
+## Excluding one or multiple facts
+
+To exclude a fact from being served to consumers in the future, you'll need access to the `fact` table.
+Setting the `exclusion_reason` field to any value will result in it being excluded from all FactStreams.
+
+For multiple facts it is recommended to apply this change within one transaction to prevent setting of the update
+trigger multiple times.
+
+```sql
+begin;
+UPDATE fact f SET exclusion_reason = 'issue-42'
+    WHERE (f.header ->> 'id') IN (
+        '924e21d0-f8f3-4162-9d18-8efd7656c494',
+        'd0ca1057-c20a-4c32-b4a3-a00523fa471e'
+    );
+commit;
+```
+
+## Exclusion prior to version X.X
+
+If you use FactCast on a version prior to X.X, you have two ways for excluding (blacklisting) facts:
 
-## The postgres blacklist _(default)_
+### The postgres blacklist _(default)_
 
 Blocked fact IDs can be added to a table named "blacklist" within the postgresDB. Inserting a new factId into the table  
 triggers a notification that is sent to the FactCast and updates the internal representations of the running Factcast
@@ -29,7 +53,7 @@ Servers to make sure that changes take immediate effect.
 In order to document why the facts have been blacklisted, you can use the reason column (of type text). It will
 not be use for anything else, so there are no expectations on the content.
 
-## The filesystem blacklist
+### The filesystem blacklist
 
 As an alternative you can provide a list of blocked fact-ids in JSON format from a file located in the classpath or the
 filesystem. Consult the [properties page](/setup/properties#blacklist) on how to set this up.

factcast-site/documentation-docsy/content/en/Setup/properties.md

@@ -53,6 +53,7 @@ description: Properties you can use to configure FactCast
 | factcast.store.enumerationDirectModeEnabled                  | Despite of a Schema-Registry being defined or not, if set to true, enumeration of types or namespace will examine the data in the store directly, so that you only see data from already published facts.                                                                                                                                                                                                                                           | false                                    |
 | factcast.store.autoFlushDelay                                | When catching up, if production of a full notification of facts takes longer than this value (in milliseconds), an additional flush is inserted into the pipelin in order to send the notification as is to the client. This is done in order to balance parallelization vs. network/compression efficiency.                                                                                                                                        | <nobr>10000</nobr>                       |
 | factcast.store.catchupStrategy                               | Available: CURSOR and CHUNKED. Cursor does the catchup query in one go and keeps the cursor open until the facts are sent to the client. Chunked runs queries limited to page-size rows instead.                                                                                                                                                                                                                                                    | CURSOR                                   |
+| factcast.store.useInternalExclusion                          | Enable only after finishing the blacklist migration described [here]({{< ref "migration.md">}}). Disables the blacklist entirely and instead enables filtering via exclusion column on the database level.                                                                                                                                                                                                                                          | false                                    |
 
 {{< alert severity="warning" size="small" >}}
 
@@ -170,7 +171,9 @@ spring.grpc.server.keep-alive.permit-time=100
 
 ---
 
-### Blacklist
+### Blacklist (deprecated in version `0.X.X`)
+
+> The following properties only take effect as long as `factcast.store.useInternalExclusion` is set to false.
 
 | Property                    | Description                                                                                                | Default                  | Example                            |
 | --------------------------- | :--------------------------------------------------------------------------------------------------------- | :----------------------- | :--------------------------------- |

factcast-store/src/main/java/org/factcast/store/StoreConfigurationProperties.java

@@ -242,6 +242,13 @@ public class StoreConfigurationProperties implements InitializingBean {
    */
   int fromScratchCatchupLogSuppressionSampleRate = 50;
 
+  /**
+   * When true, fact queries ignore excluded facts via the migrated {@code exclusion_reason} column
+   * on the fact table instead of relying on post query filtering based on a separate blacklist
+   * table. Defaults to false.
+   */
+  boolean useInternalExclusion = false;
+
   public boolean isSchemaRegistryConfigured() {
     return schemaRegistryUrl != null;
   }

factcast-store/src/main/java/org/factcast/store/internal/PgConstants.java

@@ -97,6 +97,8 @@ public class PgConstants {
 
   public static final String COLUMN_SER = "ser";
 
+  public static final String COLUMN_EXCLUSION_REASON = "exclusion_reason";
+
   public static final String COLUMN_CID = "cid";
 
   private static final String COLUMN_STATE = "state";
@@ -320,6 +322,14 @@ public class PgConstants {
   public static final String FIRST_SERIAL_AFTER_DATE =
       "SELECT MIN(firstser) FROM " + TABLE_DATE2SERIAL + " WHERE factDate >= ?";
 
+  public static String notExcludedAnd(StoreConfigurationProperties props) {
+    if (props.isUseInternalExclusion()) {
+      return COLUMN_EXCLUSION_REASON + " IS NULL AND ";
+    } else {
+      return "";
+    }
+  }
+
   private static String fromHeader(String attributeName) {
     return PgConstants.COLUMN_HEADER + "->>'" + attributeName + "' AS " + attributeName;
   }
@@ -338,6 +348,8 @@ public static String createTailIndex(
       with.append(" WITH (fastupdate = false) ");
     }
 
+    final String notExcludedAnd = notExcludedAnd(props);
+
     return "create index concurrently "
         + indexName
         + " on "
@@ -347,6 +359,7 @@ public static String createTailIndex(
         + " jsonb_path_ops) "
         + with.toString()
         + " WHERE "
+        + notExcludedAnd
         + COLUMN_SER
         + ">"
         + ser;

factcast-store/src/main/java/org/factcast/store/internal/PgFactStore.java

@@ -287,7 +287,7 @@ private State doGetState(@NotNull Collection<FactSpec> specs, long lastMatchingS
     return metrics.time(
         StoreMetrics.OP.GET_STATE_FOR,
         () -> {
-          PgQueryBuilder pgQueryBuilder = new PgQueryBuilder(specs);
+          PgQueryBuilder pgQueryBuilder = new PgQueryBuilder(specs, props.isUseInternalExclusion());
           String stateSQL = pgQueryBuilder.createStateSQL();
           PreparedStatementSetter statementSetter =
               pgQueryBuilder.createStatementSetter(new AtomicLong(lastMatchingSerial));

factcast-store/src/main/java/org/factcast/store/internal/PgFactStoreInternalConfiguration.java

@@ -260,18 +260,19 @@ public BlacklistDataProvider blacklistProvider(
       Blacklist blacklist,
       EventBus eventBus,
       JdbcTemplate jdbc,
-      BlacklistConfigurationProperties blacklistConfiguration) {
+      BlacklistConfigurationProperties blacklistConfiguration,
+      StoreConfigurationProperties storeConfiguration) {
     switch (blacklistConfiguration.getType()) {
       case POSTGRES:
-        return new PgBlacklistDataProvider(eventBus, jdbc, blacklist);
+        return new PgBlacklistDataProvider(eventBus, jdbc, blacklist, storeConfiguration);
       case RESOURCE:
         return new ResourceBasedBlacklistDataProvider(
             resourceLoader, blacklistConfiguration, blacklist);
       default:
         log.warn(
             "No Provider found for blacklist type {}. Using default postgres provider.",
             blacklistConfiguration.getType());
-        return new PgBlacklistDataProvider(eventBus, jdbc, blacklist);
+        return new PgBlacklistDataProvider(eventBus, jdbc, blacklist, storeConfiguration);
     }
   }
 

factcast-store/src/main/java/org/factcast/store/internal/PgFactStream.java

@@ -105,7 +105,8 @@ void connect() {
   @VisibleForTesting
   @NotNull
   PgSynchronizedQuery createPgSynchronizedQuery() {
-    PgQueryBuilder q = new PgQueryBuilder(request.specs(), statementHolder);
+    PgQueryBuilder q =
+        new PgQueryBuilder(request.specs(), statementHolder, props.isUseInternalExclusion());
     String sql = q.createSQL();
     log.trace("created query SQL for {} - SQL={}", request.specs(), sql);
     PreparedStatementSetter setter = q.createStatementSetter(serial);

factcast-store/src/main/java/org/factcast/store/internal/catchup/chunked/PgChunkedCatchup.java

@@ -144,7 +144,7 @@ where ser in (select ser from chunk)
   int prepareTemporaryTable(JdbcTemplate jdbc, String tempTableName) {
     createTempTable(jdbc, tempTableName);
 
-    final var b = new PgQueryBuilder(req.specs(), statementHolder);
+    final var b = new PgQueryBuilder(req.specs(), statementHolder, props.isUseInternalExclusion());
     b.useTempTable(tempTableName);
 
     final var fromSerial = new AtomicLong(Math.max(serial.get(), fastForward));

factcast-store/src/main/java/org/factcast/store/internal/catchup/cursor/PgCursorCatchup.java

@@ -72,7 +72,7 @@ public void run() {
   void fetch(JdbcTemplate jdbc) {
     jdbc.setFetchSize(props.getPageSize());
     jdbc.setQueryTimeout(0); // disable query timeout
-    final var b = new PgQueryBuilder(req.specs(), statementHolder);
+    final var b = new PgQueryBuilder(req.specs(), statementHolder, props.isUseInternalExclusion());
     final var extractor = new PgFactExtractor(serial);
     final var fromSerial = serial.get() < fastForward ? new AtomicLong(fastForward) : serial;
     final var catchupSQL = b.createSQL();

factcast-store/src/main/java/org/factcast/store/internal/filter/blacklist/BlacklistType.java

@@ -17,5 +17,5 @@
 
 public enum BlacklistType {
   POSTGRES,
-  RESOURCE
+  RESOURCE,
 }

factcast-store/src/main/java/org/factcast/store/internal/filter/blacklist/PgBlacklistDataProvider.java

@@ -21,6 +21,7 @@
 import java.util.*;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
+import org.factcast.store.StoreConfigurationProperties;
 import org.factcast.store.internal.notification.BlacklistChangeNotification;
 import org.springframework.beans.factory.*;
 import org.springframework.jdbc.core.JdbcTemplate;
@@ -31,24 +32,34 @@ public final class PgBlacklistDataProvider
   private final EventBus bus;
   private final JdbcTemplate jdbc;
   private final Blacklist blacklist;
+  private final StoreConfigurationProperties props;
 
   public PgBlacklistDataProvider(
       @NonNull EventBus eventBus,
       @NonNull JdbcTemplate jdbcTemplate,
-      @NonNull Blacklist blacklist) {
+      @NonNull Blacklist blacklist,
+      @NonNull StoreConfigurationProperties properties) {
     this.bus = eventBus;
     this.jdbc = jdbcTemplate;
     this.blacklist = blacklist;
+    this.props = properties;
   }
 
   @Override
   public void afterSingletonsInstantiated() {
     bus.register(this);
-    updateBlacklist(); // initially necessary
+    if (!props.isUseInternalExclusion()) updateBlacklist(); // initially necessary
   }
 
   @Subscribe
   public void on(BlacklistChangeNotification signal) {
+    if (props.isUseInternalExclusion()) {
+      log.warn(
+          "A change to the blacklist table was detected, but filtering uses the "
+              + "internal exclusion column (factcast.store.useInternalExclusion=true). "
+              + "While an automated sync is in place for now, exclusion only is possible via the fact table in the future.");
+      return;
+    }
     log.debug("A potential change on blacklist table was triggered.");
     updateBlacklist();
   }

factcast-store/src/main/java/org/factcast/store/internal/pipeline/ServerPipelineFactory.java

@@ -48,7 +48,10 @@ public ServerPipeline create(
         new BufferedTransformingServerPipeline(
             chain, factTransformerService, FactTransformers.createFor(subreq), maxBufferSize);
 
-    chain = new BlacklistFilterServerPipeline(chain, blacklist);
+    if (!properties.isUseInternalExclusion()) {
+      chain = new BlacklistFilterServerPipeline(chain, blacklist);
+    }
+
     chain = new AutoFlushingServerPipeline(chain, properties.getAutoFlushDelay());
     return chain;
   }

factcast-store/src/main/java/org/factcast/store/internal/query/PgQueryBuilder.java

@@ -46,17 +46,22 @@ public class PgQueryBuilder {
 
   private final @NonNull Collection<FactSpec> factSpecs;
   private final CurrentStatementHolder statementHolder;
+  private final boolean useInternalExclusion;
   private String tempTableName = null;
 
-  public PgQueryBuilder(@NonNull Collection<FactSpec> specs) {
+  public PgQueryBuilder(@NonNull Collection<FactSpec> specs, boolean useInternalExclusion) {
     factSpecs = specs;
     statementHolder = null;
+    this.useInternalExclusion = useInternalExclusion;
   }
 
   public PgQueryBuilder(
-      @NonNull Collection<FactSpec> specs, @NonNull CurrentStatementHolder holder) {
+      @NonNull Collection<FactSpec> specs,
+      @NonNull CurrentStatementHolder holder,
+      boolean useInternalExclusion) {
     factSpecs = specs;
     this.statementHolder = holder;
+    this.useInternalExclusion = useInternalExclusion;
   }
 
   public PreparedStatementSetter createStatementSetter(@NonNull AtomicLong serial) {
@@ -224,7 +229,13 @@ private String createWhereClause() {
           predicates.add(sb.toString());
         });
     String predicatesAsString = String.join(OR, predicates);
-    return "( " + predicatesAsString + " ) " + AND + PgConstants.COLUMN_SER + ">?";
+    StringBuilder sb = new StringBuilder("( ").append(predicatesAsString).append(" ) ");
+    // when internal exclusion is enabled, only ever match non-excluded facts; matches the
+    // partial GIN index idx_fact_header_active WHERE exclusion_reason IS NULL
+    if (useInternalExclusion) {
+      sb.append(AND).append(PgConstants.COLUMN_EXCLUSION_REASON).append(" IS NULL ");
+    }
+    return sb.append(AND).append(PgConstants.COLUMN_SER).append(">?").toString();
   }
 
   public String createSQL() {

factcast-store/src/main/resources/db/changelog/db.changelog-master.yaml

@@ -605,3 +605,59 @@ databaseChangeLog:
            relativeToChangelogFile: true
            splitStatements: false
            stripComments: true
+
+  - changeSet:
+      id: issue3609_add_exclusion_reason_to_facts
+      author: benediktkaiser
+      runInTransaction: false
+      comment: add nullable exclusion_reason column to fact, replacing the blacklist table
+      changes:
+        - sqlFile:
+            encoding: utf8
+            path: factcast/issue3609/add_exclusion_reason_to_facts.sql
+            relativeToChangelogFile: true
+            splitStatements: false
+            stripComments: true
+        - sqlFile:
+            encoding: utf8
+            path: factcast/issue3609/sync_blacklist_change_to_exclusion_reason.sql
+            relativeToChangelogFile: true
+            splitStatements: false
+            stripComments: true
+
+  - changeSet:
+      id: issue3609_create_blacklist_migration_procedure
+      author: benediktkaiser
+      runInTransaction: false
+      comment: create the batched, non-blocking procedure that migrates the blacklist into fact.exclusion_reason
+      changes:
+        - sqlFile:
+            encoding: utf8
+            path: factcast/issue3609/migrate_blacklist_to_exclusion_reason_procedure.sql
+            relativeToChangelogFile: true
+            splitStatements: false
+            stripComments: true
+
+  - changeSet:
+      id: issue3609_migrate_blacklist_to_exclusion_reason
+      author: benediktkaiser
+      runInTransaction: false
+      comment: execute procedure that migrates existing blacklist entries into fact.exclusion_reason in batches and create index.
+      preConditions:
+        - onFail: MARK_RAN
+        - sqlCheck:
+            expectedResult: t
+            sql: SELECT count_estimate('SELECT * FROM fact') < 10000000;
+      changes:
+        - sqlFile:
+            encoding: utf8
+            path: factcast/issue3609/call_migrate_blacklist_to_exclusion_reason_procedure.sql
+            relativeToChangelogFile: true
+            splitStatements: false
+            stripComments: true
+        - sqlFile:
+            encoding: utf8
+            path: factcast/issue3609/create_partial_idx_fact_header_active.sql
+            relativeToChangelogFile: true
+            splitStatements: false
+            stripComments: true

factcast-store/src/main/resources/db/changelog/factcast/issue3609/add_exclusion_reason_to_facts.sql

@@ -0,0 +1 @@
+alter table fact add column if not exists exclusion_reason text;