diff --git a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/bus/MessageBusChannels.java b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/bus/MessageBusChannels.java
index 8c13211c4..3ee2f4dfe 100644
--- a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/bus/MessageBusChannels.java
+++ b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/bus/MessageBusChannels.java
@@ -15,5 +15,6 @@ public enum MessageBusChannels {
RUN_VALIDATED,
CHANGE_NEW,
EXPERIMENT_RESULT_NEW,
+ SCHEMA_SYNC,
FOOBAR
}
diff --git a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/SchemaServiceImpl.java b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/SchemaServiceImpl.java
index 76ee86e62..2b1a30d1a 100644
--- a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/SchemaServiceImpl.java
+++ b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/SchemaServiceImpl.java
@@ -187,19 +187,24 @@ public Integer add(Schema schemaDTO){
.setParameter(1, schema.id).executeUpdate();
em.createNativeQuery("DELETE FROM dataset_schemas WHERE schema_id = ?1")
.setParameter(1, schema.id).executeUpdate();
- mediator.newOrUpdatedSchema(schema);
+ newOrUpdatedSchema(schema);
}
}
else {
schema.id = null;
schema.persist();
em.flush();
- mediator.newOrUpdatedSchema(schema);
+ newOrUpdatedSchema(schema);
}
log.debugf("Added schema %s (%d), URI %s", schema.name, schema.id, schema.uri);
return schema.id;
}
+ private void newOrUpdatedSchema(SchemaDAO schema) {
+ log.debugf("Push schema event for async run schemas update: %d (%s)", schema.id, schema.uri);
+ Util.registerTxSynchronization(tm, txStatus -> mediator.queueSchemaSync(schema.id));
+ }
+
private void verifyNewSchema(Schema schemaDTO) {
if (schemaDTO.uri == null || Arrays.stream(ALL_URNS).noneMatch(scheme -> schemaDTO.uri.startsWith(scheme + ":"))) {
throw ServiceException.badRequest("Please use URI starting with one of these schemes: " + Arrays.toString(ALL_URNS));
diff --git a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java
index 5514a1801..8a4c8a2e4 100644
--- a/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java
+++ b/horreum-backend/src/main/java/io/hyperfoil/tools/horreum/svc/ServiceMediator.java
@@ -9,7 +9,6 @@
import io.hyperfoil.tools.horreum.api.data.Test;
import io.hyperfoil.tools.horreum.api.services.ExperimentService;
import io.hyperfoil.tools.horreum.entity.data.ActionDAO;
-import io.hyperfoil.tools.horreum.entity.data.SchemaDAO;
import io.hyperfoil.tools.horreum.events.DatasetChanges;
import io.smallrye.reactive.messaging.annotations.Blocking;
import io.vertx.core.Vertx;
@@ -76,6 +75,10 @@ public class ServiceMediator {
@Channel("run-recalc-out")
Emitter<Integer> runEmitter;
+ @OnOverflow(value = OnOverflow.Strategy.BUFFER, bufferSize = 10000)
+ @Channel("schema-sync-out")
+ Emitter<Integer> schemaEmitter;
+
public ServiceMediator() {
}
@@ -158,6 +161,18 @@ void queueRunRecalculation(int runId) {
runEmitter.send(runId);
}
+ @Incoming("schema-sync-in")
+ @Blocking(ordered = false, value = "horreum.schema.pool")
+ @ActivateRequestContext
+ public void processSchemaSync(int schemaId) {
+ runService.onNewOrUpdatedSchema(schemaId);
+ }
+
+ @Transactional(Transactional.TxType.NOT_SUPPORTED)
+ void queueSchemaSync(int schemaId) {
+ schemaEmitter.send(schemaId);
+ }
+
void dataPointsProcessed(DataPoint.DatasetProcessedEvent event) {
experimentService.onDatapointsCreated(event);
}
@@ -206,9 +221,6 @@ void importTestToAll(TestExport test) {
subscriptionService.importSubscriptions(test);
}
- public void newOrUpdatedSchema(SchemaDAO schema) {
- runService.processNewOrUpdatedSchema(schema);
- }
public void updateFingerprints(int testId) {
datasetService.updateFingerprints(testId);
}
@@ -222,4 +234,5 @@ public void validateDataset(Integer datasetId) {
public void validateSchema(int schemaId) {
schemaService.revalidateAll(schemaId);
}
+
}
diff --git a/horreum-backend/src/main/resources/application.properties b/horreum-backend/src/main/resources/application.properties
index 881deee9c..826d79a9c 100644
--- a/horreum-backend/src/main/resources/application.properties
+++ b/horreum-backend/src/main/resources/application.properties
@@ -46,6 +46,18 @@ mp.messaging.outgoing.run-recalc-out.address=run-recalc
mp.messaging.outgoing.run-recalc-out.durable=true
mp.messaging.outgoing.run-recalc-out.container-id=horreum-broker
mp.messaging.outgoing.run-recalc-out.link-name=run-recalc
+# schema-sync incoming
+mp.messaging.incoming.schema-sync-in.connector=smallrye-amqp
+mp.messaging.incoming.schema-sync-in.address=schema-sync
+mp.messaging.incoming.schema-sync-in.durable=true
+mp.messaging.incoming.schema-sync-in.container-id=horreum-broker
+mp.messaging.incoming.schema-sync-in.link-name=schema-sync
+# schema-sync outgoing
+mp.messaging.outgoing.schema-sync-out.connector=smallrye-amqp
+mp.messaging.outgoing.schema-sync-out.address=schema-sync
+mp.messaging.outgoing.schema-sync-out.durable=true
+mp.messaging.outgoing.schema-sync-out.container-id=horreum-broker
+mp.messaging.outgoing.schema-sync-out.link-name=schema-sync
## Datasource updated by Liquibase - the same as app but always with superuser credentials
@@ -74,8 +86,9 @@ horreum.test-mode=false
#quarkus.native.additional-build-args=
# thread pool sizes
-smallrye.messaging.worker.horreum.dataset.pool.max-concurrency=10
-smallrye.messaging.worker.horreum.run.pool.max-concurrency=10
+smallrye.messaging.worker.horreum.dataset.pool.max-concurrency=7
+smallrye.messaging.worker.horreum.run.pool.max-concurrency=7
+smallrye.messaging.worker.horreum.schema.pool.max-concurrency=7
hibernate.jdbc.time_zone=UTC
diff --git a/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/SchemaServiceTest.java b/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/SchemaServiceTest.java
index fb6a378ce..d1861eafa 100644
--- a/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/SchemaServiceTest.java
+++ b/horreum-backend/src/test/java/io/hyperfoil/tools/horreum/svc/SchemaServiceTest.java
@@ -30,6 +30,8 @@
import io.quarkus.test.junit.TestProfile;
import io.quarkus.test.oidc.server.OidcWiremockTestResource;
import io.restassured.common.mapper.TypeRef;
+import jakarta.inject.Inject;
+import jakarta.persistence.Tuple;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.TestInfo;
@@ -256,7 +258,7 @@ private void testExportImport(boolean wipe) {
}
@org.junit.jupiter.api.Test
- public void testFindUsages(TestInfo info) throws InterruptedException {
+ public void testFindUsages() throws InterruptedException {
Test test = createTest(createExampleTest("nofilter"));
createComparisonSchema();
uploadExampleRuns(test);
@@ -268,12 +270,145 @@ public void testFindUsages(TestInfo info) throws InterruptedException {
assertNotEquals(0, report.data.size());
- List<SchemaService.LabelLocation> usages = jsonRequest().get("/api/schema/findUsages?label=".concat("category"))
- .then().statusCode(200).extract().body().as(List.class);
+ List<SchemaService.LabelLocation> usages =
+ jsonRequest().get("/api/schema/findUsages?label=".concat("category"))
+ .then().statusCode(200).extract().body().as(List.class);
assertNotNull(usages);
+ }
+
+ @org.junit.jupiter.api.Test
+ public void testCreateSchemaAfterRunWithArrayData() throws InterruptedException {
+ String schemaUri = "urn:unknown:schema";
+ Test test = createTest(createExampleTest("dummy-test"));
+
+ ArrayNode data = JsonNodeFactory.instance.arrayNode();
+ data.addObject().put("$schema", schemaUri).put("foo", "bar");
+ data.addObject().put("$schema", schemaUri).put("foo", "bar");
+ int runId = uploadRun(data.toString(), test.name);
+ assertTrue(runId > 0);
+
+ // no validation errors
+ assertEquals(0, em.createNativeQuery("SELECT COUNT(*)::::int FROM run_validationerrors").getSingleResult());
+ assertEquals(0, em.createNativeQuery("SELECT COUNT(*)::::int FROM dataset_validationerrors").getSingleResult());
+
+ List<?> runSchemasBefore = em.createNativeQuery("SELECT * FROM run_schemas WHERE runid = ?1").setParameter(1, runId).getResultList();
+ assertEquals(0, runSchemasBefore.size());
+
+ // create the schema afterward
+ Schema schema = createSchema("Unknown schema", schemaUri);
+ assertNotNull(schema);
+ assertTrue(schema.id > 0);
+
+ TestUtil.eventually(() -> {
+ Util.withTx(tm, () -> {
+ em.clear();
+ List<?> runSchemas = em.createNativeQuery("SELECT * FROM run_schemas WHERE runid = ?1").setParameter(1, runId).getResultList();
+ // two records as the run is an array of two objects, both referencing the same schema
+ assertEquals(2, runSchemas.size());
+ return null;
+ });
+ });
+ }
+
+ @org.junit.jupiter.api.Test
+ public void testCreateSchemaAfterRunWithMultipleSchemas() throws InterruptedException {
+ String firstSchemaUri = "urn:unknown1:schema";
+ String secondSchemaUri = "urn:unknown2:schema";
+ Test test = createTest(createExampleTest("dummy-test"));
+ ArrayNode data = JsonNodeFactory.instance.arrayNode();
+ data.addObject().put("$schema", firstSchemaUri).put("foo", "bar");
+ data.addObject().put("$schema", secondSchemaUri).put("foo", "zip");
+ int runId = uploadRun(data.toString(), test.name);
+ assertTrue(runId > 0);
+
+ // no validation errors
+ assertEquals(0, em.createNativeQuery("SELECT COUNT(*)::::int FROM run_validationerrors").getSingleResult());
+ assertEquals(0, em.createNativeQuery("SELECT COUNT(*)::::int FROM dataset_validationerrors").getSingleResult());
+
+ List<?> runSchemasBefore = em.createNativeQuery("SELECT * FROM run_schemas WHERE runid = ?1").setParameter(1, runId).getResultList();
+ assertEquals(0, runSchemasBefore.size());
+
+ // create the schema 1 afterward
+ Schema schema1 = createSchema("Unknown schema 1", firstSchemaUri);
+ assertNotNull(schema1);
+ assertTrue(schema1.id > 0);
+
+ TestUtil.eventually(() -> {
+ Util.withTx(tm, () -> {
+ em.clear();
+ List<Tuple> runSchemas = em.createNativeQuery("SELECT * FROM run_schemas WHERE runid = ?1", Tuple.class).setParameter(1, runId).getResultList();
+ // 1 record as the run is an array of two objects referencing different schemas and only the first one is created
+ assertEquals(1, runSchemas.size());
+ assertEquals(schema1.id, (int)runSchemas.get(0).get(3));
+ return null;
+ });
+ });
}
+ @org.junit.jupiter.api.Test
+ public void testCreateSchemaAfterRunWithObjectData() throws InterruptedException {
+ String schemaUri = "urn:unknown:schema";
+ Test test = createTest(createExampleTest("dummy-test"));
+ ObjectNode data = JsonNodeFactory.instance.objectNode();
+ data.put("$schema", schemaUri).put("foo", "bar");
+ int runId = uploadRun(data.toString(), test.name);
+ assertTrue(runId > 0);
+
+ // no validation errors
+ assertEquals(0, em.createNativeQuery("SELECT COUNT(*)::::int FROM run_validationerrors").getSingleResult());
+ assertEquals(0, em.createNativeQuery("SELECT COUNT(*)::::int FROM dataset_validationerrors").getSingleResult());
+
+ List<?> runSchemasBefore = em.createNativeQuery("SELECT * FROM run_schemas WHERE runid = ?1").setParameter(1, runId).getResultList();
+ assertEquals(0, runSchemasBefore.size());
+
+ // create the schema afterward
+ Schema schema = createSchema("Unknown schema", schemaUri);
+ assertNotNull(schema);
+ assertTrue(schema.id > 0);
+
+ TestUtil.eventually(() -> {
+ Util.withTx(tm, () -> {
+ em.clear();
+ List<?> runSchemas = em.createNativeQuery("SELECT * FROM run_schemas WHERE runid = ?1").setParameter(1, runId).getResultList();
+ // run has single object data, thus referencing one schema
+ assertEquals(1, runSchemas.size());
+ return null;
+ });
+ });
+ }
+
+ @org.junit.jupiter.api.Test
+ public void testChangeUriForReferencedSchema() throws InterruptedException {
+ String schemaUri = "urn:dummy:schema";
+ Schema schema = createSchema("Dummy schema", schemaUri);
+ assertNotNull(schema);
+ assertTrue(schema.id > 0);
+
+ Test test = createTest(createExampleTest("dummy-test"));
+
+ ArrayNode data = JsonNodeFactory.instance.arrayNode();
+ data.addObject().put("$schema", schemaUri).put("foo", "bar");
+ data.addObject().put("$schema", schemaUri).put("foo", "bar");
+ int runId = uploadRun(data.toString(), test.name);
+ assertTrue(runId > 0);
+
+ // no validation errors
+ assertEquals(0, em.createNativeQuery("SELECT COUNT(*)::::int FROM run_validationerrors").getSingleResult());
+ assertEquals(0, em.createNativeQuery("SELECT COUNT(*)::::int FROM dataset_validationerrors").getSingleResult());
+
+ List<?> runSchemasBefore = em.createNativeQuery("SELECT * FROM run_schemas WHERE runid = ?1").setParameter(1, runId).getResultList();
+ assertEquals(2, runSchemasBefore.size());
+
+ // update the schema uri afterward
+ schema.uri = "urn:new-dummy:schema";
+ Schema updatedSchema = addOrUpdateSchema(schema);
+ assertNotNull(updatedSchema);
+ assertEquals(schema.id, updatedSchema.id);
+
+ List<?> runSchemasAfter = em.createNativeQuery("SELECT * FROM run_schemas WHERE runid = ?1").setParameter(1, runId).getResultList();
+ assertEquals(0, runSchemasAfter.size());
+ }
}
diff --git a/horreum-web/src/Banner.tsx b/horreum-web/src/Banner.tsx
index cbf96c549..cf5e5ec82 100644
--- a/horreum-web/src/Banner.tsx
+++ b/horreum-web/src/Banner.tsx
@@ -3,19 +3,58 @@ import { useEffect, useState } from "react"
import { Alert } from "@patternfly/react-core"
import {Banner as BannerData, bannerApi} from "./api"
+function getAlertBanner(banner: BannerData) {
+ return (
+ <Alert variant={banner.severity as any} title={banner.title} isInline>
+ <div dangerouslySetInnerHTML={{ __html: banner.message || "" }}></div>
+ </Alert>
+ )
+}
+
+// 30 seconds
+const DEFAULT_TIMEOUT = 30000
+export type TimeoutBannerProps = {
+ bannerData: BannerData
+ timeout?: number
+ onTimeout?: () => void
+}
+
+export function TimeoutBanner({bannerData, timeout, onTimeout}: TimeoutBannerProps) {
+ timeout = timeout ?? DEFAULT_TIMEOUT
+ const [banner, setBanner] = useState<BannerData | undefined>(bannerData)
+
+ useEffect(() => {
+ if (banner) {
+ const timeoutId = setTimeout(() => {
+ setBanner(undefined)
+ if (onTimeout) {
+ onTimeout()
+ }
+ }, timeout)
+
+ return () => clearTimeout(timeoutId);
+ }
+ }, [])
+
+ if (!banner) {
+ return null
+ }
+
+ return getAlertBanner(banner)
+}
+
export default function Banner() {
const [banner, setBanner] = useState<BannerData>()
const [updateCounter, setUpdateCounter] = useState(0)
useEffect(() => {
- setTimeout(() => setUpdateCounter(updateCounter + 1), 60000)
+ const timeoutId = setTimeout(() => setUpdateCounter(updateCounter + 1), 60000)
bannerApi.get().then(setBanner)
+
+ return () => clearTimeout(timeoutId)
}, [updateCounter])
if (!banner) {
return null
}
- return (
- <Alert variant={banner.severity as any} title={banner.title} isInline>
- <div dangerouslySetInnerHTML={{ __html: banner.message || "" }}></div>
- </Alert>
- )
+
+ return getAlertBanner(banner)
}
diff --git a/horreum-web/src/domain/schemas/Schema.tsx b/horreum-web/src/domain/schemas/Schema.tsx
index d87f99ddb..736952263 100644
--- a/horreum-web/src/domain/schemas/Schema.tsx
+++ b/horreum-web/src/domain/schemas/Schema.tsx
@@ -37,10 +37,11 @@ import SavedTabs, { SavedTab, TabFunctions, modifiedFunc, resetFunc, saveFunc }
import TeamSelect from "../../components/TeamSelect"
import Transformers from "./Transformers"
import Labels from "./Labels"
-import {Access, getSchema, Schema as SchemaDef, schemaApi} from "../../api"
+import {Access, getSchema, Schema as SchemaDef, schemaApi, Banner as BannerData} from "../../api"
import SchemaExportImport from "./SchemaExportImport"
import {AppContext} from "../../context/appContext";
import {AppContextType} from "../../context/@types/appContextTypes";
+import { TimeoutBanner, TimeoutBannerProps } from "../../Banner"
type SchemaParams = {
schemaId: string
@@ -212,6 +213,7 @@ export default function Schema() {
const [editorSchema, setEditorSchema] = useState(schema?.schema ? toString(schema?.schema) : undefined)
const [modifiedSchema, setModifiedSchema] = useState(schema)
const [modified, setModified] = useState(false)
+ const [showMessageBanner, setShowMessageBanner] = useState(false)
// any tester can save to add new labels/transformers
const isTester = useTester()
@@ -246,8 +248,6 @@ export default function Schema() {
return schemaUri || undefined
}
-
-
const save = () => {
if (!modified) {
return Promise.resolve(schemaIdVal)
@@ -261,13 +261,29 @@ export default function Schema() {
return schemaApi.add(newSchema)
.then(id=> id,
error => alerting.dispatchError(error, "SAVE_SCHEMA", "Failed to save schema")
- ).then(id => setSchemaIdVal(id))
+ ).then(id => {
+ setSchemaIdVal(id)
+ setShowMessageBanner(true)
+ })
}
const transformersFuncsRef = useRef<TabFunctions>()
const labelsFuncsRef = useRef<TabFunctions>()
+ const bannerProps = {
+ bannerData: {
+ title: "Schema sync alert",
+ message: "Horreum schema changes is processed asynchronously. As a result, changes in runs and schemas links may not be reflected immediately.",
+ severity: "info",
+ active: true,
+ },
+ onTimeout: () => setShowMessageBanner(false)
+ } as TimeoutBannerProps
+
return (
<PageSection>
+ {showMessageBanner && (
+ <TimeoutBanner {...bannerProps} />
+ )}
{loading && (
<Bullseye>
<Spinner />