Skip to content
Merged
Show file tree
Hide file tree
Changes from 12 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,15 @@ BatchResult buildBatch() throws BatchValidationException
{
lastPrefetchRowNumber = asInteger(_delegate.get(0));
Object keyObj = pkSupplier.get();
Object key = jdbcType.convert(keyObj);
Object key;
try
{
key = jdbcType.convert(keyObj);
}
catch (Exception e)
{
throw new IllegalArgumentException(keyFieldName + " value not valid on row " + lastPrefetchRowNumber);
}

if (numericKey)
{
Expand Down
131 changes: 126 additions & 5 deletions api/src/org/labkey/api/exp/api/SampleChangeNotify.java
Original file line number Diff line number Diff line change
Expand Up @@ -16,41 +16,162 @@
package org.labkey.api.exp.api;

import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.labkey.api.admin.notification.NotificationService;
import org.labkey.api.data.AccumulatingCommitTask;
import org.labkey.api.data.Container;
import org.labkey.api.data.DbScope;
import org.labkey.api.security.SecurityManager;
import org.labkey.api.security.User;
import org.labkey.api.security.permissions.Permission;
import org.labkey.api.security.permissions.ReadPermission;
import org.labkey.api.util.logging.LogHelper;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;

public enum SampleChangeNotify
{
SampleDataChanged;

private static final Logger LOG = LogHelper.getLogger(SampleChangeNotify.class, "Sample data change WebSocket notifications");
private static final List<SampleCountsListener> LISTENERS = new CopyOnWriteArrayList<>();

/** Commit task that dedupes per container, so repeated registrations in one transaction notify once. */
public record FireSampleDataChangedTask(@Nullable Container container) implements Runnable
// Stands for "any sample type"; sample type row ids are positive
private static final Long ANY_SAMPLE_TYPE = -1L;

/** Told which sample types' samples changed in a container. Empty sampleTypeIds means any sample type may have. */
public interface SampleCountsListener
{
/** Called inside the changing transaction just before it commits, or right away when there is no transaction */
void samplesChanging(@NotNull Container container, @NotNull Set<Long> sampleTypeIds);

/**
* Called after the change commits
* @return true if the listener will send {@link #sendSampleDataChangedEvent} itself once its counts are current
*/
boolean samplesChanged(@NotNull Container container, @NotNull Set<Long> sampleTypeIds);
}

public static void addListener(SampleCountsListener listener)
{
LISTENERS.add(listener);
}

/*
* The commit tasks below are AccumulatingCommitTasks: every registration in a transaction contributes its "values"
* to one set per container, which the task then processes once. Here a value is a sample type row id, and callers
* and listeners both speak in those ids, with an empty set meaning "any sample type in this folder may have
* changed". The two methods below convert between that vocabulary and what the task accumulates, because an empty
* set contributes nothing to accumulate and so would vanish instead of widening the set -- ANY_SAMPLE_TYPE carries
* it as a value that does accumulate. It exists only between these two methods; no caller or listener sees it.
*/

/** Caller's ids -> values to accumulate. Nothing nameable becomes the sentinel, so it survives accumulation. */
private static Collection<Long> sampleTypeIdsToTaskValues(@Nullable Collection<Long> sampleTypeIds)
{
return sampleTypeIds == null || sampleTypeIds.isEmpty() ? Set.of(ANY_SAMPLE_TYPE) : sampleTypeIds;
}

/**
* Accumulated values -> ids for the listener. One sentinel widens the whole set: once some registration in the
* transaction couldn't name its types, the ids the others contributed are no longer a complete list, so the
* listener is told "any type here" rather than a subset it would otherwise trust.
*/
private static Set<Long> taskValuesToSampleTypeIds(Set<Long> values)
{
return values.contains(ANY_SAMPLE_TYPE) ? Set.of() : values;
}

/**
* Tells listeners about a change inside the current transaction, once per container. Use this alone only when the
* caller also arranges a post-commit {@link #fireSampleDataChanged}; otherwise use {@link SampleDataChangedTask}.
Comment thread
labkey-jeckels marked this conversation as resolved.
Resolved
*/
public static void recordSampleDataChange(@NotNull DbScope scope, @NotNull Container container, @Nullable Collection<Long> sampleTypeIds)
{
new SampleDataChangingTask(container).register(scope, sampleTypeIdsToTaskValues(sampleTypeIds), DbScope.CommitTaskOption.PRECOMMIT);
}

private static class SampleDataChangingTask extends AccumulatingCommitTask<Container, Long>
{
SampleDataChangingTask(@NotNull Container container)
{
super(container);
}

/**
* Deliberately lets a listener failure out. This runs inside the changing transaction, so a failed write has
* already aborted it and every later statement in it will fail too -- there is no version of carrying on that
* commits the sample change and merely loses the bookkeeping. Failing here rolls the whole thing back, which
* the caller can report, rather than committing and leaving counts that silently disagree with the samples.
*/
@Override
protected void process(@NotNull Container container, @NotNull Set<Long> values)
{
for (SampleCountsListener listener : LISTENERS)
listener.samplesChanging(container, taskValuesToSampleTypeIds(values));
}
}

/** Commit task that collects the changed sample type ids per container, so a transaction notifies once per container. */
public static class SampleDataChangedTask extends AccumulatingCommitTask<Container, Long>
{
private final Container _container;

public SampleDataChangedTask(@NotNull Container container)
{
super(container);
_container = container;
}

/** Records the change in the current transaction and notifies after it commits; null or empty sampleTypeIds means any sample type in the container. */
public void register(@NotNull DbScope scope, @Nullable Collection<Long> sampleTypeIds)
{
recordSampleDataChange(scope, _container, sampleTypeIds);
register(scope, sampleTypeIdsToTaskValues(sampleTypeIds), DbScope.CommitTaskOption.POSTCOMMIT);
}

@Override
public void run()
protected void process(@NotNull Container container, @NotNull Set<Long> sampleTypeIds)
{
fireSampleDataChanged(container);
fireSampleDataChanged(container, taskValuesToSampleTypeIds(sampleTypeIds));
}
}

public static void fireSampleDataChanged(@Nullable Container container)
/** After a change commits, tells listeners which sample types changed, then notifies clients unless a listener will once its counts are current. */
public static void fireSampleDataChanged(@Nullable Container container, @Nullable Collection<Long> sampleTypeIds)
{
if (container == null)
return;

Set<Long> ids = sampleTypeIds == null ? Set.of() : Set.copyOf(sampleTypeIds);
boolean deferred = false;
for (SampleCountsListener listener : LISTENERS)
{
// Unlike samplesChanging, this is past the commit: a failure here costs a refresh notification and
// nothing else, so it mustn't stop the other listeners being told
try
{
deferred |= listener.samplesChanged(container, ids);
}
catch (Exception e)
{
LOG.warn("Sample counts listener failed to handle a change in " + container.getPath(), e);
}
}

if (!deferred)
sendSampleDataChangedEvent(container);
}

/** Sends the SampleDataChanged WebSocket event to every user who can read the container */
public static void sendSampleDataChangedEvent(@NotNull Container container)
{
try
{
Set<Class<? extends Permission>> read = Collections.singleton(ReadPermission.class);
Expand Down
56 changes: 56 additions & 0 deletions api/src/org/labkey/api/exp/api/SampleCountsService.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
/*
* Copyright (c) 2026 LabKey Corporation
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.labkey.api.exp.api;

import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.labkey.api.data.ContainerFilter;
import org.labkey.api.data.DbSchema;
import org.labkey.api.data.SQLFragment;
import org.labkey.api.services.ServiceRegistry;

/**
* Serves sample counts that a module maintains ahead of time, so a caller can read them instead of counting samples.
* Registered by the module that owns those counts; null when no module does, in which case callers count live.
*/
public interface SampleCountsService
{
static void setInstance(SampleCountsService impl)
{
ServiceRegistry.get().registerService(SampleCountsService.class, impl);
}

@Nullable
static SampleCountsService get()
{
return ServiceRegistry.get().getService(SampleCountsService.class);
}

/**
* A scalar expression for how many samples a sample type has in the folders the container filter allows, suitable
* for a correlated subquery against a sample type table.
*
* @param sampleTypeRowId the enclosing query's sample type RowId column
* @param liveCount counts the samples directly; the returned expression falls back to it wherever the maintained
* counts can't answer, so a count that isn't ready is slow rather than wrong. An implementation
* may embed it more than once, and must append this fragment itself at each position rather than
* copying its SQL text, so that its bound parameters stay aligned with their placeholders.
* @return null when this service can't answer at all, leaving the caller to use liveCount
*/
@Nullable
SQLFragment getSampleCountSQL(@NotNull SQLFragment sampleTypeRowId, @NotNull ContainerFilter cf,
@NotNull DbSchema schema, @NotNull SQLFragment liveCount);
}
50 changes: 42 additions & 8 deletions assay/src/org/labkey/assay/AssayExperimentListener.java
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@
*/
package org.labkey.assay;

import org.jetbrains.annotations.NotNull;
import org.labkey.api.data.AccumulatingCommitTask;
import org.labkey.api.data.Container;
import org.labkey.api.data.DbScope;
import org.labkey.api.data.DbScope.CommitTaskOption;
Expand All @@ -26,16 +28,51 @@
import org.labkey.api.security.User;

import java.util.List;
import java.util.Set;

public class AssayExperimentListener implements ExperimentListener
{
/*
* Both deletes below fire once per item, so a folder delete reaches them once per batch or run it holds. Collecting
* the items and de-indexing them together keeps that to one commit task, which is also what the deindex calls want
* -- each resolves its items to document ids and deletes them in a single search request.
*
* Keyed by a constant rather than by container: a document id is not container-scoped, and taking no container is
* what lets these run after the container itself is gone.
*/

private static class DeindexAssayBatchesTask extends AccumulatingCommitTask<String, ExpExperiment>

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@labkey-jeckels I included some unrelated AccumulatingCommitTask switch in this branch to address the warnings logged while running tests.

{
private DeindexAssayBatchesTask()
{
super(DeindexAssayBatchesTask.class.getName());
}

@Override
protected void process(@NotNull String key, @NotNull Set<ExpExperiment> batches)
{
AssayManager.get().deindexAssayBatches(batches);
}
}

private static class DeindexAssayRunsTask extends AccumulatingCommitTask<String, ExpRun>
{
private DeindexAssayRunsTask()
{
super(DeindexAssayRunsTask.class.getName());
}

@Override
protected void process(@NotNull String key, @NotNull Set<ExpRun> runs)
{
AssayManager.get().deindexAssayRuns(runs);
}
}

@Override
public void afterExperimentDeleted(Container c, User user, ExpExperiment experiment)
{
DbScope.getLabKeyScope().addCommitTask(() ->
AssayManager.get().deindexAssayBatches(List.of(experiment)),
CommitTaskOption.POSTCOMMIT
);
new DeindexAssayBatchesTask().register(DbScope.getLabKeyScope(), List.of(experiment), CommitTaskOption.POSTCOMMIT);
}

@Override
Expand All @@ -47,10 +84,7 @@ public void afterExperimentSaved(Container c, User user, ExpExperiment experimen
@Override
public void afterRunDelete(ExpProtocol protocol, ExpRun run, User user)
{
DbScope.getLabKeyScope().addCommitTask(() ->
AssayManager.get().deindexAssayRuns(List.of(run)),
CommitTaskOption.POSTCOMMIT
);
new DeindexAssayRunsTask().register(DbScope.getLabKeyScope(), List.of(run), CommitTaskOption.POSTCOMMIT);
}

@Override
Expand Down
Loading
Loading