ldap based pwnotify

server/src/main/java/password/pwm/http/servlet/admin/AdminServlet.java

@@ -710,7 +710,7 @@ public class AdminServlet extends ControlledPwmServlet
     }
 
     @ActionHandler( action = "readPwNotifyStatus" )
-    public ProcessStatus restreadPwNotifyStatus( final PwmRequest pwmRequest ) throws IOException, DatabaseException, PwmUnrecoverableException
+    public ProcessStatus restreadPwNotifyStatus( final PwmRequest pwmRequest ) throws IOException, PwmUnrecoverableException
     {
         int key = 0;
         if ( !pwmRequest.getConfig().readSettingAsBoolean( PwmSetting.PW_EXPY_NOTIFY_ENABLE ) )
@@ -722,31 +722,6 @@ public class AdminServlet extends ControlledPwmServlet
             return ProcessStatus.Halt;
         }
 
-
-        {
-            ErrorInformation errorInformation = null;
-            try
-            {
-                if ( !pwmRequest.getPwmApplication().getDatabaseService().getAccessor().isConnected() )
-                {
-                    errorInformation = new ErrorInformation( PwmError.ERROR_DB_UNAVAILABLE, "database is not connected" );
-                }
-            }
-            catch ( PwmUnrecoverableException e )
-            {
-                errorInformation = e.getErrorInformation();
-            }
-
-            if ( errorInformation != null )
-            {
-                final DisplayElement displayElement = new DisplayElement( String.valueOf( key++ ), DisplayElement.Type.string, "Status",
-                        "Database must be functioning to view Password Notify status.  Current database error: "
-                                + errorInformation.toDebugStr() );
-                pwmRequest.outputJsonResult( RestResultBean.withData( new PwNotifyStatusBean( Collections.singletonList( displayElement ), false ) ) );
-                return ProcessStatus.Halt;
-            }
-        }
-
         final List<DisplayElement> statusData = new ArrayList<>( );
         final Configuration config = pwmRequest.getConfig();
         final Locale locale = pwmRequest.getLocale();
@@ -780,8 +755,11 @@ public class AdminServlet extends ControlledPwmServlet
                         "Last Job Duration", TimeDuration.between( storedJobState.getLastStart(), storedJobState.getLastCompletion() ).asLongString( locale ) ) );
             }
 
-            statusData.add( new DisplayElement( String.valueOf( key++ ), DisplayElement.Type.string,
-                    "Last Job Server Instance",  storedJobState.getServerInstance() ) );
+            if ( !StringUtil.isEmpty( storedJobState.getServerInstance() ) )
+            {
+                statusData.add( new DisplayElement( String.valueOf( key++ ), DisplayElement.Type.string,
+                        "Last Job Server Instance", storedJobState.getServerInstance() ) );
+            }
 
             if ( storedJobState.getLastError() != null )
             {

server/src/main/java/password/pwm/svc/pwnotify/PwNotifyDbStorageService.java

@@ -37,6 +37,8 @@ import password.pwm.util.java.StringUtil;
 
 class PwNotifyDbStorageService implements PwNotifyStorageService
 {
+    private static final String DB_STATE_STRING = "PwNotifyJobState";
+
     private static final DatabaseTable TABLE = DatabaseTable.PW_NOTIFY;
     private final PwmApplication pwmApplication;
 
@@ -46,7 +48,7 @@ class PwNotifyDbStorageService implements PwNotifyStorageService
     }
 
     @Override
-    public StoredNotificationState readStoredState(
+    public StoredNotificationState readStoredUserState(
             final UserIdentity userIdentity,
             final SessionLabel sessionLabel
     )
@@ -79,7 +81,7 @@ class PwNotifyDbStorageService implements PwNotifyStorageService
         return JsonUtil.deserialize( rawDbValue, StoredNotificationState.class );
     }
 
-    public void writeStoredState(
+    public void writeStoredUserState(
             final UserIdentity userIdentity,
             final SessionLabel sessionLabel,
             final StoredNotificationState storedNotificationState
@@ -110,4 +112,39 @@ class PwNotifyDbStorageService implements PwNotifyStorageService
             throw new PwmUnrecoverableException( new ErrorInformation( PwmError.ERROR_DB_UNAVAILABLE, e.getMessage() ) );
         }
     }
+
+    @Override
+    public StoredJobState readStoredJobState()
+            throws PwmUnrecoverableException
+    {
+        try
+        {
+            final String strValue = pwmApplication.getDatabaseService().getAccessor().get( DatabaseTable.PW_NOTIFY, DB_STATE_STRING );
+            if ( StringUtil.isEmpty( strValue ) )
+            {
+                return new StoredJobState( null, null, null, null, false );
+            }
+            return JsonUtil.deserialize( strValue, StoredJobState.class );
+        }
+        catch ( DatabaseException e )
+        {
+            throw new PwmUnrecoverableException( new ErrorInformation( PwmError.ERROR_DB_UNAVAILABLE, e.getMessage() ) );
+        }
+    }
+
+    @Override
+    public void writeStoredJobState( final StoredJobState storedJobState )
+            throws PwmUnrecoverableException
+    {
+        try
+        {
+            final String strValue = JsonUtil.serialize( storedJobState );
+            pwmApplication.getDatabaseService().getAccessor().put( DatabaseTable.PW_NOTIFY, DB_STATE_STRING, strValue );
+        }
+        catch ( DatabaseException e )
+        {
+            throw new PwmUnrecoverableException( new ErrorInformation( PwmError.ERROR_DB_UNAVAILABLE, e.getMessage() ) );
+        }
+    }
+
 }

server/src/main/java/password/pwm/svc/pwnotify/PwNotifyEngine.java

@@ -35,7 +35,6 @@ import password.pwm.error.PwmUnrecoverableException;
 import password.pwm.ldap.LdapOperationsHelper;
 import password.pwm.ldap.UserInfo;
 import password.pwm.ldap.UserInfoFactory;
-import password.pwm.svc.stats.EventRateMeter;
 import password.pwm.svc.stats.Statistic;
 import password.pwm.svc.stats.StatisticsManager;
 import password.pwm.util.LocaleHelper;
@@ -47,16 +46,18 @@ import password.pwm.util.macro.MacroMachine;
 
 import java.io.IOException;
 import java.io.Writer;
-import java.math.BigDecimal;
 import java.time.Duration;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
-import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Locale;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
 
 public class PwNotifyEngine
 {
@@ -69,14 +70,14 @@ public class PwNotifyEngine
     private final Writer debugWriter;
     private final StringBuffer internalLog = new StringBuffer(  );
     private final List<UserPermission> permissionList;
+    private final PwNotifyStorageService storageService;
+    private final Supplier<Boolean> cancelFlag;
 
     private final ConditionalTaskExecutor debugOutputTask = new ConditionalTaskExecutor(
             this::periodicDebugOutput,
             new ConditionalTaskExecutor.TimeDurationPredicate( 1, TimeDuration.Unit.MINUTES )
     );
 
-    private EventRateMeter eventRateMeter = new EventRateMeter( TimeDuration.of( 5, TimeDuration.Unit.MINUTES ) );
-
     private final AtomicInteger examinedCount = new AtomicInteger( 0 );
     private final AtomicInteger noticeCount = new AtomicInteger( 0 );
     private Instant startTime;
@@ -85,10 +86,14 @@ public class PwNotifyEngine
 
     PwNotifyEngine(
             final PwmApplication pwmApplication,
+            final PwNotifyStorageService storageService,
+            final Supplier<Boolean> cancelFlag,
             final Writer debugWriter
     )
     {
         this.pwmApplication = pwmApplication;
+        this.cancelFlag = cancelFlag;
+        this.storageService = storageService;
         this.settings = PwNotifySettings.fromConfiguration( pwmApplication.getConfig() );
         this.debugWriter = debugWriter;
         this.permissionList = pwmApplication.getConfig().readSettingAsUserPermission( PwmSetting.PW_EXPY_NOTIFY_PERMISSION );
@@ -133,7 +138,7 @@ public class PwNotifyEngine
             internalLog.delete( 0, internalLog.length() );
             running = true;
 
-            if ( !canRunOnThisServer() )
+            if ( !canRunOnThisServer() || cancelFlag.get() )
             {
                 return;
             }
@@ -156,36 +161,22 @@ public class PwNotifyEngine
             );
 
             log( "ldap search complete, examining users..." );
+
+            final ThreadPoolExecutor threadPoolExecutor = createExecutor( pwmApplication );
             while ( workQueue.hasNext() )
             {
-                if ( !checkIfRunningOnMaster() )
+                if ( !checkIfRunningOnMaster() || cancelFlag.get() )
                 {
                     final String msg = "job interrupted, server is no longer the cluster master.";
                     log( msg );
                     throw PwmUnrecoverableException.newException( PwmError.ERROR_SERVICE_NOT_AVAILABLE, msg );
                 }
 
-                checkIfRunningOnMaster(  );
-
-                final List<UserIdentity> batch = new ArrayList<>(  );
-                final int batchSize = settings.getBatchCount();
-
-                while ( batch.size() < batchSize && workQueue.hasNext() )
-                {
-                    batch.add( workQueue.next() );
-                }
+                threadPoolExecutor.submit( new ProcessJob( workQueue.next() ) );
+            }
 
-                final Instant startBatch = Instant.now();
-                processBatch( batch );
-                eventRateMeter.markEvents( batchSize );
-                final TimeDuration batchTime = TimeDuration.fromCurrent( startBatch );
-                final TimeDuration pauseTime = TimeDuration.of(
-                        settings.getBatchTimeMultiplier().multiply( new BigDecimal( batchTime.asMillis() ) ).longValue(),
-                        TimeDuration.Unit.MILLISECONDS );
-                pauseTime.pause();
+            JavaHelper.closeAndWaitExecutor( threadPoolExecutor, TimeDuration.DAY );
 
-                debugOutputTask.conditionallyExecuteTask();
-            }
             log( "job complete, " + examinedCount + " users evaluated in " + TimeDuration.fromCurrent( startTime ).asCompactString()
                     + ", sent " + noticeCount + " notices."
             );
@@ -198,26 +189,46 @@ public class PwNotifyEngine
 
     private void periodicDebugOutput()
     {
-        log( "job in progress, " + examinedCount + " users evaluated in "
+        final String msg = "job in progress, " + examinedCount + " users evaluated in "
                 + TimeDuration.fromCurrent( startTime ).asCompactString()
-                + ", sent " + noticeCount + " notices."
-        );
+                + ", sent " + noticeCount + " notices.";
+        log( msg );
     }
 
-    private void processBatch( final Collection<UserIdentity> batch )
-            throws PwmUnrecoverableException
+    private class ProcessJob implements Runnable
     {
-        for ( final UserIdentity userIdentity : batch )
+        final UserIdentity userIdentity;
+
+        ProcessJob( final UserIdentity userIdentity )
         {
-            processUserIdentity( userIdentity );
+            this.userIdentity = userIdentity;
+        }
+
+        @Override
+        public void run()
+        {
+            try
+            {
+                processUserIdentity( userIdentity );
+                debugOutputTask.conditionallyExecuteTask();
+            }
+            catch ( Exception e )
+            {
+                LOGGER.trace( "unexpected error processing user '" + userIdentity.toDisplayString() + "', error: " + e.getMessage() );
+            }
         }
     }
 
-    private boolean processUserIdentity(
+    private void processUserIdentity(
             final UserIdentity userIdentity
     )
             throws PwmUnrecoverableException
     {
+        if ( !canRunOnThisServer() || cancelFlag.get() )
+        {
+            return;
+        }
+
         examinedCount.incrementAndGet();
         final ChaiUser theUser = pwmApplication.getProxiedChaiUser( userIdentity );
         final Instant passwordExpirationTime = LdapOperationsHelper.readPasswordExpirationTime( theUser );
@@ -225,36 +236,31 @@ public class PwNotifyEngine
         if ( passwordExpirationTime == null )
         {
             LOGGER.trace( SESSION_LABEL, "skipping user '" + userIdentity.toDisplayString() + "', has no password expiration" );
-            return false;
+            return;
         }
 
         if ( passwordExpirationTime.isBefore( Instant.now() ) )
         {
             LOGGER.trace( SESSION_LABEL, "skipping user '" + userIdentity.toDisplayString() + "', password expiration is in the past" );
-            return false;
+            return;
         }
 
         final int nextDayInterval = figureNextDayInterval( passwordExpirationTime );
         if ( nextDayInterval < 1 )
         {
             LOGGER.trace( SESSION_LABEL, "skipping user '" + userIdentity.toDisplayString() + "', password expiration time is not within an interval" );
-            return false;
+            return;
         }
 
         if ( checkIfNoticeAlreadySent( userIdentity, passwordExpirationTime, nextDayInterval ) )
         {
             log( "notice for interval " + nextDayInterval + " already sent for " + userIdentity.toDisplayString() );
-            return false;
+            return;
         }
 
         log( "sending notice to " + userIdentity.toDisplayString() + " for interval " + nextDayInterval );
-        {
-            final PwNotifyDbStorageService dbStorage = new PwNotifyDbStorageService( pwmApplication );
-            dbStorage.writeStoredState( userIdentity, SESSION_LABEL, new StoredNotificationState( passwordExpirationTime, Instant.now(), nextDayInterval ) );
-        }
-
+        storageService.writeStoredUserState( userIdentity, SESSION_LABEL, new StoredNotificationState( passwordExpirationTime, Instant.now(), nextDayInterval ) );
         sendNoticeEmail( userIdentity );
-        return true;
     }
 
     private int figureNextDayInterval(
@@ -288,8 +294,7 @@ public class PwNotifyEngine
     )
             throws PwmUnrecoverableException
     {
-        final PwNotifyDbStorageService dbStorage = new PwNotifyDbStorageService( pwmApplication );
-        final StoredNotificationState storedState = dbStorage.readStoredState( userIdentity, SESSION_LABEL );
+        final StoredNotificationState storedState = storageService.readStoredUserState( userIdentity, SESSION_LABEL );
 
         if ( storedState == null )
         {
@@ -365,4 +370,20 @@ public class PwNotifyEngine
 
         LOGGER.trace( SessionLabel.PWNOTIFY_SESSION_LABEL, output );
     }
+
+    private ThreadPoolExecutor createExecutor( final PwmApplication pwmApplication )
+    {
+        final ThreadFactory threadFactory = JavaHelper.makePwmThreadFactory( JavaHelper.makeThreadName( pwmApplication, this.getClass() ), true );
+        final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
+                1,
+                10,
+                1,
+                TimeUnit.MINUTES,
+                new LinkedBlockingDeque<>(),
+                threadFactory
+        );
+        threadPoolExecutor.allowCoreThreadTimeOut( true );
+        return threadPoolExecutor;
+    }
+
 }

server/src/main/java/password/pwm/svc/pwnotify/PwNotifyLdapStorageService.java

@@ -0,0 +1,157 @@
+/*
+ * Password Management Servlets (PWM)
+ * http://www.pwm-project.org
+ *
+ * Copyright (c) 2006-2009 Novell, Inc.
+ * Copyright (c) 2009-2018 The PWM Project
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+
+package password.pwm.svc.pwnotify;
+
+import com.novell.ldapchai.ChaiUser;
+import com.novell.ldapchai.exception.ChaiOperationException;
+import com.novell.ldapchai.exception.ChaiUnavailableException;
+import com.novell.ldapchai.util.ConfigObjectRecord;
+import password.pwm.PwmApplication;
+import password.pwm.bean.SessionLabel;
+import password.pwm.bean.UserIdentity;
+import password.pwm.config.profile.LdapProfile;
+import password.pwm.error.ErrorInformation;
+import password.pwm.error.PwmError;
+import password.pwm.error.PwmUnrecoverableException;
+import password.pwm.util.java.JsonUtil;
+import password.pwm.util.java.StringUtil;
+
+class PwNotifyLdapStorageService implements PwNotifyStorageService
+{
+    private final PwmApplication pwmApplication;
+    private final PwNotifySettings settings;
+
+    private enum CoreType
+    {
+        User( "0007" ),
+        ProxyUser( "0005" ),;
+
+        private final String recordID;
+
+        CoreType( final String recordID )
+        {
+            this.recordID = recordID;
+        }
+
+        public String getRecordID()
+        {
+            return recordID;
+        }
+    }
+
+    PwNotifyLdapStorageService( final PwmApplication pwmApplication, final PwNotifySettings settings )
+    {
+        this.pwmApplication = pwmApplication;
+        this.settings = settings;
+    }
+
+    @Override
+    public StoredNotificationState readStoredUserState(
+            final UserIdentity userIdentity,
+            final SessionLabel sessionLabel
+    )
+            throws PwmUnrecoverableException
+    {
+        final ConfigObjectRecord configObjectRecord = getUserCOR( userIdentity, CoreType.User );
+        final String payload = configObjectRecord.getPayload();
+        if ( StringUtil.isEmpty( payload ) )
+        {
+            return JsonUtil.deserialize( payload, StoredNotificationState.class );
+        }
+        return null;
+    }
+
+    public void writeStoredUserState(
+            final UserIdentity userIdentity,
+            final SessionLabel sessionLabel,
+            final StoredNotificationState storedNotificationState
+    )
+            throws PwmUnrecoverableException
+    {
+        final ConfigObjectRecord configObjectRecord = getUserCOR( userIdentity, CoreType.User );
+        final String payload = JsonUtil.serialize( storedNotificationState );
+        try
+        {
+            configObjectRecord.updatePayload( payload );
+        }
+        catch ( ChaiOperationException e )
+        {
+            final String msg = "error writing user pwNotifyStatus attribute '" + settings.getLdapUserAttribute() + ", error: " + e.getMessage();
+            final ErrorInformation errorInformation = new ErrorInformation( PwmError.ERROR_LDAP_DATA_ERROR, msg );
+            throw new PwmUnrecoverableException( errorInformation );
+        }
+        catch ( ChaiUnavailableException e )
+        {
+            throw PwmUnrecoverableException.fromChaiException( e );
+        }
+    }
+
+    @Override
+    public StoredJobState readStoredJobState()
+            throws PwmUnrecoverableException
+    {
+        final LdapProfile ldapProfile = pwmApplication.getConfig().getDefaultLdapProfile();
+        final UserIdentity proxyUser = ldapProfile.getProxyUser( pwmApplication );
+        final ConfigObjectRecord configObjectRecord = getUserCOR( proxyUser, CoreType.ProxyUser );
+        final String payload = configObjectRecord.getPayload();
+
+        if ( StringUtil.isEmpty( payload ) )
+        {
+            return new StoredJobState( null, null, null, null, false );
+        }
+        return JsonUtil.deserialize( payload, StoredJobState.class );
+    }
+
+    @Override
+    public void writeStoredJobState( final StoredJobState storedJobState )
+            throws PwmUnrecoverableException
+    {
+        final LdapProfile ldapProfile = pwmApplication.getConfig().getDefaultLdapProfile();
+        final UserIdentity proxyUser = ldapProfile.getProxyUser( pwmApplication );
+        final ConfigObjectRecord configObjectRecord = getUserCOR( proxyUser, CoreType.ProxyUser );
+        final String payload = JsonUtil.serialize( storedJobState );
+
+        try
+        {
+            configObjectRecord.updatePayload( payload );
+        }
+        catch ( ChaiOperationException e )
+        {
+            final String msg = "error writing user pwNotifyStatus attribute on proxy user '" + settings.getLdapUserAttribute() + ", error: " + e.getMessage();
+            final ErrorInformation errorInformation = new ErrorInformation( PwmError.ERROR_LDAP_DATA_ERROR, msg );
+            throw new PwmUnrecoverableException( errorInformation );
+        }
+        catch ( ChaiUnavailableException e )
+        {
+            throw PwmUnrecoverableException.fromChaiException( e );
+        }
+    }
+
+    private ConfigObjectRecord getUserCOR( final UserIdentity userIdentity, final CoreType coreType )
+            throws PwmUnrecoverableException
+    {
+        final String userAttr = settings.getLdapUserAttribute();
+        final ChaiUser chaiUser = pwmApplication.getProxiedChaiUser( userIdentity );
+        return ConfigObjectRecord.createNew( chaiUser, userAttr, coreType.getRecordID(), null, null );
+    }
+}

server/src/main/java/password/pwm/svc/pwnotify/PwNotifyService.java

@@ -34,10 +34,7 @@ import password.pwm.health.HealthRecord;
 import password.pwm.svc.PwmService;
 import password.pwm.svc.stats.Statistic;
 import password.pwm.svc.stats.StatisticsManager;
-import password.pwm.util.db.DatabaseException;
-import password.pwm.util.db.DatabaseTable;
 import password.pwm.util.java.JavaHelper;
-import password.pwm.util.java.JsonUtil;
 import password.pwm.util.java.StringUtil;
 import password.pwm.util.java.TimeDuration;
 import password.pwm.util.logging.PwmLogger;
@@ -45,22 +42,23 @@ import password.pwm.util.logging.PwmLogger;
 import java.time.Duration;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ExecutorService;
 
 public class PwNotifyService implements PwmService
 {
     private static final PwmLogger LOGGER = PwmLogger.forClass( PwNotifyService.class );
 
-    private ScheduledExecutorService executorService;
+    private ExecutorService executorService;
     private PwmApplication pwmApplication;
     private STATUS status = STATUS.NEW;
     private PwNotifyEngine engine;
     private PwNotifySettings settings;
     private Instant nextExecutionTime;
+    private PwNotifyStorageService storageService;
+    private ErrorInformation lastError;
 
     @Override
     public STATUS status( )
@@ -68,22 +66,10 @@ public class PwNotifyService implements PwmService
         return status;
     }
 
-    private static final String DB_STATE_STRING = "PwNotifyJobState";
 
-    private StoredJobState readStoredJobState()
-            throws PwmUnrecoverableException, DatabaseException
+    public StoredJobState getJobState() throws PwmUnrecoverableException
     {
-        final String strValue = pwmApplication.getDatabaseService().getAccessor().get( DatabaseTable.PW_NOTIFY, DB_STATE_STRING );
-        if ( StringUtil.isEmpty( strValue ) )
-        {
-            return new StoredJobState( null, null, null, null, false );
-        }
-        return JsonUtil.deserialize( strValue, StoredJobState.class );
-    }
-
-    public StoredJobState getJobState() throws DatabaseException, PwmUnrecoverableException
-    {
-        return readStoredJobState();
+        return storageService.readStoredJobState();
     }
 
     public boolean isRunning()
@@ -93,18 +79,17 @@ public class PwNotifyService implements PwmService
 
     public String debugLog()
     {
-        if ( engine != null )
+        if ( engine != null && !StringUtil.isEmpty( engine.getDebugLog() ) )
         {
             return engine.getDebugLog();
         }
-        return "";
-    }
 
-    private void writeStoredJobState( final StoredJobState storedJobState )
-            throws PwmUnrecoverableException, DatabaseException
-    {
-        final String strValue = JsonUtil.serialize( storedJobState );
-        pwmApplication.getDatabaseService().getAccessor().put( DatabaseTable.PW_NOTIFY, DB_STATE_STRING, strValue );
+        if ( lastError != null )
+        {
+            return lastError.toDebugStr();
+        }
+
+        return "";
     }
 
     @Override
@@ -120,17 +105,16 @@ public class PwNotifyService implements PwmService
         }
 
         settings = PwNotifySettings.fromConfiguration( pwmApplication.getConfig() );
-        engine = new PwNotifyEngine( pwmApplication, null );
+        storageService = new PwNotifyDbStorageService( pwmApplication );
+        //storageService = new PwNotifyLdapStorageService( pwmApplication, settings );
+        engine = new PwNotifyEngine( pwmApplication, storageService, () -> status == STATUS.CLOSED, null );
 
-        executorService = Executors.newSingleThreadScheduledExecutor(
-                JavaHelper.makePwmThreadFactory(
-                        JavaHelper.makeThreadName( pwmApplication, this.getClass() ) + "-",
-                        true
-                ) );
+        executorService = JavaHelper.makeBackgroundExecutor( pwmApplication, this.getClass() );
 
-        executorService.scheduleWithFixedDelay( new PwNotifyJob(), 1, 1, TimeUnit.MINUTES );
+        pwmApplication.scheduleFixedRateJob( new PwNotifyJob(), executorService, TimeDuration.MINUTE, TimeDuration.MINUTE );
 
         status = STATUS.OPEN;
+
     }
 
     public Instant getNextExecutionTime( )
@@ -151,9 +135,10 @@ public class PwNotifyService implements PwmService
         }
     }
 
-    private Instant figureNextJobExecutionTime() throws DatabaseException, PwmUnrecoverableException
+    private Instant figureNextJobExecutionTime()
+            throws PwmUnrecoverableException
     {
-        final StoredJobState storedJobState = readStoredJobState();
+        final StoredJobState storedJobState = storageService.readStoredJobState();
         if ( storedJobState != null )
         {
             // never run, or last job not successful.
@@ -195,23 +180,31 @@ public class PwNotifyService implements PwmService
             return Collections.emptyList();
         }
 
+        final List<HealthRecord> returnRecords = new ArrayList<>(  );
+
+        if ( lastError != null )
+        {
+            returnRecords.add( HealthRecord.forMessage( HealthMessage.PwNotify_Failure, lastError.toDebugStr() ) );
+        }
+
         try
         {
-            final StoredJobState storedJobState = readStoredJobState();
+            final StoredJobState storedJobState = storageService.readStoredJobState();
             if ( storedJobState != null )
             {
                 final ErrorInformation errorInformation = storedJobState.getLastError();
                 if ( errorInformation != null )
                 {
-                    return Collections.singletonList( HealthRecord.forMessage( HealthMessage.PwNotify_Failure, errorInformation.toDebugStr() ) );
+                    returnRecords.add( HealthRecord.forMessage( HealthMessage.PwNotify_Failure, errorInformation.toDebugStr() ) );
                 }
             }
         }
-        catch ( DatabaseException | PwmUnrecoverableException e  )
+        catch ( PwmUnrecoverableException e  )
         {
             LOGGER.error( SessionLabel.PWNOTIFY_SESSION_LABEL, "error while generating health information: " + e.getMessage() );
         }
-        return null;
+
+        return returnRecords;
     }
 
     @Override
@@ -231,7 +224,7 @@ public class PwNotifyService implements PwmService
         if ( !isRunning() )
         {
             nextExecutionTime = Instant.now();
-            executorService.schedule( new PwNotifyJob(), 1, TimeUnit.SECONDS );
+            pwmApplication.scheduleFutureJob( new PwNotifyJob(), executorService, TimeDuration.ZERO );
         }
     }
 
@@ -272,16 +265,17 @@ public class PwNotifyService implements PwmService
 
         private void doJob( )
         {
+            lastError = null;
             final Instant start = Instant.now();
             try
             {
-                writeStoredJobState( new StoredJobState( Instant.now(), null, pwmApplication.getInstanceID(), null, false ) );
+                storageService.writeStoredJobState( new StoredJobState( Instant.now(), null, pwmApplication.getInstanceID(), null, false ) );
                 StatisticsManager.incrementStat( pwmApplication, Statistic.PWNOTIFY_JOBS );
                 engine.executeJob();
 
                 final Instant finish = Instant.now();
                 final StoredJobState storedJobState = new StoredJobState( start, finish, pwmApplication.getInstanceID(), null, true );
-                writeStoredJobState( storedJobState );
+                storageService.writeStoredJobState( storedJobState );
             }
             catch ( Exception e )
             {
@@ -301,14 +295,15 @@ public class PwNotifyService implements PwmService
 
                 try
                 {
-                    writeStoredJobState( storedJobState );
+                    storageService.writeStoredJobState( storedJobState );
                 }
                 catch ( Exception e2 )
                 {
                     //no hope
                 }
                 StatisticsManager.incrementStat( pwmApplication, Statistic.PWNOTIFY_JOB_ERRORS );
-                LOGGER.debug( SessionLabel.PWNOTIFY_SESSION_LABEL, "error executing scheduled job: " + e.getMessage() );
+                LOGGER.debug( SessionLabel.PWNOTIFY_SESSION_LABEL, errorInformation );
+                lastError = errorInformation;
             }
         }
     }

server/src/main/java/password/pwm/svc/pwnotify/PwNotifySettings.java

@@ -37,7 +37,7 @@ import java.util.List;
 
 @Value
 @Builder
-public class PwNotifySettings implements Serializable
+class PwNotifySettings implements Serializable
 {
     private final List<Integer> notificationIntervals;
     private final TimeDuration maximumSkipWindow;
@@ -46,6 +46,8 @@ public class PwNotifySettings implements Serializable
     private final int batchCount;
     private final BigDecimal batchTimeMultiplier;
 
+    private final String ldapUserAttribute;
+
     static PwNotifySettings fromConfiguration( final Configuration configuration )
     {
         final PwNotifySettingsBuilder builder = PwNotifySettings.builder();
@@ -66,6 +68,9 @@ public class PwNotifySettings implements Serializable
         builder.batchTimeMultiplier( new BigDecimal( configuration.readAppProperty( AppProperty.PWNOTIFY_BATCH_DELAY_TIME_MULTIPLIER ) ) );
         builder.maximumSkipWindow( TimeDuration.of(
                 Long.parseLong( configuration.readAppProperty( AppProperty.PWNOTIFY_MAX_SKIP_RERUN_WINDOW_SECONDS ) ), TimeDuration.Unit.SECONDS ) );
+
+        builder.ldapUserAttribute( "carLicense" );
+
         return builder.build();
     }
 }

server/src/main/java/password/pwm/svc/pwnotify/PwNotifyStorageService.java

@@ -29,11 +29,17 @@ import password.pwm.error.PwmUnrecoverableException;
 interface PwNotifyStorageService
 {
 
-    StoredNotificationState readStoredState(
+    StoredNotificationState readStoredUserState(
             UserIdentity userIdentity,
             SessionLabel sessionLabel
     )
             throws PwmUnrecoverableException;
 
-    void writeStoredState( UserIdentity userIdentity, SessionLabel sessionLabel, StoredNotificationState storedNotificationState ) throws PwmUnrecoverableException;
+    void writeStoredUserState( UserIdentity userIdentity, SessionLabel sessionLabel, StoredNotificationState storedNotificationState ) throws PwmUnrecoverableException;
+
+    StoredJobState readStoredJobState()
+            throws PwmUnrecoverableException;
+
+    void writeStoredJobState( StoredJobState storedJobState )
+                    throws PwmUnrecoverableException;
 }

server/src/main/java/password/pwm/util/java/ConditionalTaskExecutor.java

@@ -26,7 +26,8 @@ import password.pwm.util.logging.PwmLogger;
 
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
-import java.util.function.Predicate;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Supplier;
 
 /**
  * <p>Executes a predefined task if a conditional has occurred.  Both the task and the conditional must be supplied by the caller.
@@ -41,35 +42,44 @@ public class ConditionalTaskExecutor
     private static final PwmLogger LOGGER = PwmLogger.forClass( ConditionalTaskExecutor.class );
 
     private Runnable task;
-    private Predicate predicate;
+    private Supplier<Boolean> predicate;
+    private final ReentrantLock lock = new ReentrantLock();
 
     /**
      * Execute the task if the conditional has been met.  Exceptions when running the task will be logged but not returned.
      */
     public void conditionallyExecuteTask( )
     {
-        if ( predicate.test( null ) )
+        lock.lock();
+        try
         {
-            try
+            if ( predicate.get() )
             {
-                task.run();
-            }
-            catch ( Throwable t )
-            {
-                LOGGER.warn( "unexpected error executing conditional task: " + t.getMessage(), t );
-            }
+                try
+                {
+                    task.run();
+                }
+                catch ( Throwable t )
+                {
+                    LOGGER.warn( "unexpected error executing conditional task: " + t.getMessage(), t );
+                }
 
+            }
+        }
+        finally
+        {
+            lock.unlock();
         }
     }
 
-    public ConditionalTaskExecutor( final Runnable task, final Predicate predicate )
+    public ConditionalTaskExecutor( final Runnable task, final Supplier<Boolean> predicate )
     {
         this.task = task;
         this.predicate = predicate;
     }
 
 
-    public static class TimeDurationPredicate implements Predicate
+    public static class TimeDurationPredicate implements Supplier<Boolean>
     {
         private final TimeDuration timeDuration;
         private volatile Instant nextExecuteTimestamp;
@@ -93,7 +103,7 @@ public class ConditionalTaskExecutor
         }
 
         @Override
-        public boolean test( final Object o )
+        public Boolean get()
         {
             if ( Instant.now().isAfter( nextExecuteTimestamp ) )
             {

server/src/main/java/password/pwm/util/localdb/XodusLocalDB.java

@@ -100,7 +100,7 @@ public class XodusLocalDB implements LocalDBProvider
     private final Map<LocalDB.DB, Store> cachedStoreObjects = new HashMap<>();
 
     private final ConditionalTaskExecutor outputLogExecutor = new ConditionalTaskExecutor(
-            ( ) -> outputStats(), new ConditionalTaskExecutor.TimeDurationPredicate( STATS_OUTPUT_INTERVAL ).setNextTimeFromNow( 1, TimeDuration.Unit.MINUTES )
+            ( ) -> outputStats(), new ConditionalTaskExecutor.TimeDurationPredicate( STATS_OUTPUT_INTERVAL ).setNextTimeFromNow( TimeDuration.MINUTE )
     );
 
     private BindMachine bindMachine = new BindMachine( BindMachine.DEFAULT_ENABLE_COMPRESSION, BindMachine.DEFAULT_MIN_COMPRESSION_LENGTH );