Payment operation listener firing off multiple times for single operation

Question

I have implemented a payment listener in my mainActivity. It is working, except that the majority of the time it is firing off multiple times (usually 2 or 3 times) for a single payment receipt.

Below is the method that contains the event listener. It is called from my onCreate method.
You will see that I tried to get around potnetially creating multiple instances of PaymentsRequestBuilder by making a global variable PaymentsRequestBuilder paymentsRequest; and checking that it is null at the start of the method if (paymentsRequest == null). This didn't work.

I have also established that this is not being called multiple times from onCreate, and I also tested saving the PagingToken locally as well as my database in case that was the issue, but it didn't help.

Every now and then, it will only fire once, which is making debugging more frustrating.

I had previously had it that this method was called from another method that is only called on login. The reason I moved it to onCreate was because I lost the listener too easily and then had to log out and back in to reestablish it.

I'm also using Firebase, in case that is relevant.

private void StellarExternalTransferListener (String StellarID, final String userID, final String StellarPagingToken, final String secretSeed, final Integer piclCount){
            Server server = new Server("https://horizon-testnet.stellar.org");
            final KeyPair account = KeyPair.fromAccountId(StellarID);
    if (paymentsRequest == null) {
        // Create an API call to query payments involving the account.
        paymentsRequest = server.payments().forAccount(account);

        // If some payments have already been handled, start the results from the
        // last seen payment.
        if (StellarPagingToken != null) {
            paymentsRequest.cursor(StellarPagingToken);
        }

        // `stream` will send each recorded payment, one by one, then keep the
        // connection open and continue to send you new payments as they occur.


        paymentsRequest.stream(new EventListener<OperationResponse>() {
            @Override
            public void onEvent(OperationResponse payment) {
                // Record the paging token so we can start from here next time.
                userRef.child(userID).child("StellarPagingToken").setValue(payment.getPagingToken());

                //Transaction records in the Firebase DB will be handled upon Stellar outgoing transaction success. Hence, incoming Stellar
                //transactions will only trigger a balance update, and not be responsible for creating records.
                Log.d(TAG, "StellarPagingToken: " + payment.getPagingToken());

                // The payments stream includes both sent and received payments. We only
                // want to process received payments here.
                if (payment instanceof PaymentOperationResponse) {
                    String receivedStellarID = ((PaymentOperationResponse) payment).getTo().getAccountId();

                    Log.d(TAG, "Known account check: " + account.getAccountId() + " | " + receivedStellarID);
                    if (!receivedStellarID.equals(account.getAccountId())) {

                        Log.d(TAG, "Known account: " + account.getAccountId());
                    } else {

                        final String senderStellarID = ((PaymentOperationResponse) payment).getFrom().getAccountId();
                        final String amount = ((PaymentOperationResponse) payment).getAmount();
                        final String asset = ((PaymentOperationResponse) payment).getAsset().getType();

                        Log.d(TAG, "External receipt!: " + senderStellarID + " | " + amount + " | " + asset);

                        stellarIDs.child(senderStellarID).addListenerForSingleValueEvent(new ValueEventListener() {
                            @Override
                            public void onDataChange(DataSnapshot dataSnapshot) {
                                Log.d(TAG, "dataSnapshot: " + dataSnapshot);
                                if (!dataSnapshot.exists() & asset.equals("native")) {

                                    DatabaseReference usersPicls = userRef.child(userID).child("PickleCount");
                                    onInboundTransactionConfirm(userID, usersPicls, senderStellarID, PiclConvertedLumenAmount, secretSeed);

                                }
                            }

                            @Override
                            public void onCancelled(@NonNull DatabaseError databaseError) {
                            }
                        });
                    }
                }
            }
        });
    }
}

Answer 1

Can't say why the event is fired more than once, though I have a solution for you. Try to use event deduplication to overcome the problem.

1. Cursor-based approach

Currently you save the most recent paging_token as a StellarPagingToken. When new payment arrives, you can simply check if the value of paging_token is larger than the value stored in StellarPagingToken. Value smaller or equal to the previous paging_token means that the event has been already processed and you can safely discard it.

While this approach is quite simple, it guarantees 100% payment events processing only if events arrive in strict succession. I don't know how the event streaming is implemented in Horizon. Maybe there is a non-zero probability of events "racing", so...

2. Processed entries set approach

You can maintain a set of recently processed events. In practice, it can be a primitive LRU cache, containing, say, 20 recently processed payments (actually, there is no need to retain anything except the paging_token itself). When new payment arrives, you just check if our cache contains a paging_token. If yes, skip further processing. Otherwise, add new paging_token to the cache and proceed. It is a more sophisticated solution, but it requires additional computational resources, therefore it's not recommended for performance-critical code (however, I don't think that your code will process hundreds of payments per second, so it won't significantly slow down the system performance).

Off-topic. The problem with event firing more than once is one of the major headaches in distributed queue-based systems. Most queue implementations can't guarantee strictly-only-once event delivery for configurations with more than one queue broker.

Answer 2

It may well be that firebase is relevant. Firebase does not have strong consistency. If you are relying on Firebase to get the value for StellarPagingToken you might experience the symptoms you're describing. Have you tried using a local copy of StellarPagingToken and fetching it from Firebase only when necessary?

Answer 3

After 2 frustrating days, I think I have it. I eliminated Firebase as the culprit because I am using a singleValueEventListner to retrieve the token.

I then took the route suggested by @Orbit_Lens, to store the last known StellarPagingToken locally and compare it against each new StellarPagingToken. I had already attempted this by using a global variable, but this time I went withSharedPreferences`. This didn't work.

I then noticed that the times for the repeat operations where within milliseconds of one another, sometimes even equal to the millisecond. This made me think that the issue may not be with multiple events of the same operation firing off, but rather that one event was for some reason resulting in multiple executions of itself (I don't know how this event stream works, so I was just making guesses at this stage). I then tried to suppress this by wrapping everything within the onEvent method in an AtomicBoolean class. I did this with the SharedPreferences stuff commented out. This didn't work.

Finally I ran it with the AtomicBoolean and the SharedPreferences stuff included. This worked! I have inserted both into the same code from my question (see below).

It doesn't feel like a comfortable solution, but it has stood up to a multiple tests, without one duplicate transaction being posted. I have also tested the creation of multiple transactions and then opening the app, to confirm that the issue does not arise when the stream handles multiple events on startup. The solution still held.

In short, I feel like a have applied enough 'duck tape' to the issue for now, but would still be open to a more robust solution, should one exist.

        paymentsRequest.stream(new EventListener<OperationResponse>() {
            @Override
            public void onEvent(OperationResponse payment) {
AtomicBoolean hasRun = new AtomicBoolean(false);
                    if (hasRun.compareAndSet(false, true)) {
                // Record the paging token so we can start from here next time.
                userRef.child(userID).child("StellarPagingToken").setValue(payment.getPagingToken());

                //Transaction records in the Firebase DB will be handled upon Stellar outgoing transaction success. Hence, incoming Stellar
                //transactions will only trigger a balance update, and not be responsible for creating records.
                Log.d(TAG, "StellarPagingToken: " + payment.getPagingToken());

                // The payments stream includes both sent and received payments. We only
                // want to process received payments here.
                if (payment instanceof PaymentOperationResponse) {
                    String receivedStellarID = ((PaymentOperationResponse) payment).getTo().getAccountId();

                    Log.d(TAG, "Known account check: " + account.getAccountId() + " | " + receivedStellarID);
                    if (!receivedStellarID.equals(account.getAccountId())) {

                        Log.d(TAG, "Known account: " + account.getAccountId());
                    } else {

final SharedPreferences StellarToken = getSharedPreferences(
                                        "stellar_token", MODE_PRIVATE);
                                String savedStellarToken = StellarToken.getString("stellar_token", "Boo");

                                if (!savedStellarToken.equals(tokenTracker)) {

                                    SharedPreferences.Editor StellarTokenEditor = StellarToken.edit();
                                    StellarTokenEditor.putString("stellar_token", payment.getPagingToken());
                                    StellarTokenEditor.apply();

                        final String senderStellarID = ((PaymentOperationResponse) payment).getFrom().getAccountId();
                        final String amount = ((PaymentOperationResponse) payment).getAmount();
                        final String asset = ((PaymentOperationResponse) payment).getAsset().getType();

                        Log.d(TAG, "External receipt!: " + senderStellarID + " | " + amount + " | " + asset);

                        stellarIDs.child(senderStellarID).addListenerForSingleValueEvent(new ValueEventListener() {
                            @Override
                            public void onDataChange(DataSnapshot dataSnapshot) {
                                Log.d(TAG, "dataSnapshot: " + dataSnapshot);
                                if (!dataSnapshot.exists() & asset.equals("native")) {

                                    DatabaseReference usersPicls = userRef.child(userID).child("PickleCount");
                                    onInboundTransactionConfirm(userID, usersPicls, senderStellarID, PiclConvertedLumenAmount, secretSeed);

                                }
                            }

                            @Override
                            public void onCancelled(@NonNull DatabaseError databaseError) {
                            }
                        });
                      }
                    }
              }  }
            }