Feat/submit (#271)

* improve submit error message

* added manual witness receipts to submit unit test

* simplify the witness receipts testing.

* witness receipts accounted for and submit operation completes
---------

Signed-off-by: 2byrds <[email protected]>

src/keria/app/agenting.py

@@ -309,11 +309,13 @@ def __init__(self, hby, rgy, agentHab, agency, caid, **opts):
         self.exchanges = decking.Deck()
         self.grants = decking.Deck()
         self.admits = decking.Deck()
+        self.submits = decking.Deck()
 
         receiptor = agenting.Receiptor(hby=hby)
         self.witq = agenting.WitnessInquisitor(hby=self.hby)
         self.witPub = agenting.WitnessPublisher(hby=self.hby)
         self.witDoer = agenting.WitnessReceiptor(hby=self.hby)
+        self.witSubmitDoer = agenting.WitnessReceiptor(hby=self.hby, force=True)
 
         self.rep = storing.Respondant(hby=hby, cues=self.cues, mbx=Mailboxer(name=self.hby.name, temp=self.hby.temp))
 
@@ -341,8 +343,9 @@ def __init__(self, hby, rgy, agentHab, agency, caid, **opts):
         self.exc = exchanging.Exchanger(hby=hby, handlers=handlers)
         grouping.loadHandlers(exc=self.exc, mux=self.mux)
         protocoling.loadHandlers(hby=self.hby, exc=self.exc, notifier=self.notifier)
+        self.submitter = Submitter(hby=hby, submits=self.submits, witRec=self.witSubmitDoer)
         self.monitor = longrunning.Monitor(hby=hby, swain=self.swain, counselor=self.counselor, temp=hby.temp,
-                                           registrar=self.registrar, credentialer=self.credentialer, exchanger=self.exc)
+                                           registrar=self.registrar, credentialer=self.credentialer, submitter=self.submitter, exchanger=self.exc)
 
         self.rvy = routing.Revery(db=hby.db, cues=self.cues)
         self.kvy = eventing.Kevery(db=hby.db,
@@ -386,6 +389,7 @@ def __init__(self, hby, rgy, agentHab, agency, caid, **opts):
             SeekerDoer(seeker=self.seeker, cues=self.verifier.cues, tock=self.tocks.get("seeker", 0.0)),
             ExchangeCueDoer(seeker=self.exnseeker, cues=self.exc.cues, queries=self.queries,
                             tock=self.tocks.get("exchangecue", 0.0)),
+            self.submitter,
         ])
 
         super(Agent, self).__init__(doers=doers, always=True, **opts)
@@ -1232,3 +1236,45 @@ def on_post(req, rep):
         rep.status = falcon.HTTP_202
         rep.content_type = "application/json"
         rep.data = op.to_json().encode("utf-8")
+
+class Submitter(doing.DoDoer):
+    def __init__(self, hby, submits, witRec):
+        """
+        Process to re-submit the last event from the KEL to the witnesses for receipts and to propogate it to each witness
+        """
+        self.hby = hby
+        self.submits = submits
+        self.witRec = witRec
+
+        super(Submitter, self).__init__(always=True)
+
+    def recur(self, tyme, deeds=None):
+        """Processes submit reqests submitting any on the cue"""
+        if self.submits:
+            msg = self.submits.popleft()
+            alias = msg["alias"]
+            hab = self.hby.habByName(name=alias)
+            sn = hab.kever.sn
+            if hab and hab.kever.wits:
+                auths = {}
+                if hasattr(msg, "code"):
+                    code = msg["code"]
+                    if code:
+                        for wit in hab.kever.wits:
+                            auths[wit] = f"{code}#{helping.nowIso8601()}"
+                witDoer = self.witRec
+                witDoer.force = True
+                self.extend([witDoer])
+                print("Re-submit waiting for witness receipts...")
+                witDoer.msgs.append(dict(pre=hab.pre, sn=sn))
+
+        else:
+            for doer in self.doers:
+                if doer.cues:
+                    cue = doer.cues.popleft()
+
+                    if len(doer.cues) == 0:
+                        print("Re-submit received all witness receipts for", cue["pre"])
+                        self.doers.remove(doer)
+
+        return super(Submitter, self).recur(tyme, deeds)

src/keria/app/aiding.py

@@ -34,6 +34,7 @@ def loadEnds(app, agency, authn):
     aidEnd = IdentifierResourceEnd()
     app.add_route("/identifiers/{name}", aidEnd)
     app.add_route("/identifiers/{name}/events", aidEnd)
+    app.add_route("/identifiers/{name}/submit", aidEnd)
 
     aidOOBIsEnd = IdentifierOOBICollectionEnd()
     app.add_route("/identifiers/{name}/oobis", aidOOBIsEnd)
@@ -756,7 +757,7 @@ def on_post(self, req, rep, name):
 
         ---
         summary: Process identifier events.
-        description: This endpoint handles the 'rot' or 'ixn' events of an identifier based on the provided request.
+        description: This endpoint handles the 'rot' or 'ixn' events of an identifier, or the request to resubmit the KEL, based on the provided request.
         tags:
         - Identifier
         parameters:
@@ -778,11 +779,16 @@ def on_post(self, req, rep, name):
                     ixn:
                       type: object
                       description: The interaction event details.
+                    submit:
+                      type: object
+                      description: The request to resubmit event details to witnesses.
                   oneOf:
                   - required:
                     - rot
                   - required:
                     - ixn
+                  - required:
+                    - submit
         responses:
             200:
               description: Successfully processed the identifier's event.
@@ -798,6 +804,8 @@ def on_post(self, req, rep, name):
                 op = self.rotate(agent, name, body)
             elif body.get("ixn") is not None:
                 op = self.interact(agent, name, body)
+            elif body.get("submit") is not None:
+                op = self.submit_id(agent, name, body)
             else:
                 raise falcon.HTTPBadRequest(
                     title="invalid request",
@@ -956,7 +964,23 @@ def interact(agent, name, body):
             metadata=dict(response=serder.ked),
         )
         return op
+
+
+    @staticmethod
+    def submit_id(agent, name, body):
+        hab = agent.hby.habByName(name)
+        if hab is None:
+            raise falcon.HTTPNotFound(title=f"No AID {name} found")
+
+        code = body.get("code")
+
+        if hab.kever.wits:
+            agent.submits.append(dict(alias=name,code=code))
+            op = agent.monitor.submit(hab.kever.prefixer.qb64, longrunning.OpTypes.submit,
+                                      metadata=dict(alias=name,sn=hab.kever.sn))
+            return op
 
+        raise falcon.HTTPBadRequest(title=f"invalid identifier submitted, {name} has no witnesses")
 
 def info(hab, rm, full=False):
     data = dict(

src/keria/app/grouping.py

@@ -187,6 +187,7 @@ def on_post(req, rep, name):
                                                     " signing member list must contain a local identifier'")
 
         hab = agent.hby.joinSignifyGroupHab(gid, name=name, mhab=mhab, smids=smids, rmids=rmids)
+
         try:
             hab.make(serder=serder, sigers=sigers)
         except (ValueError) as e:

src/keria/core/longrunning.py

@@ -20,12 +20,12 @@
 from keria.app import delegating
 
 # long running operation types
-Typeage = namedtuple("Tierage", 'oobi witness delegation group query registry credential endrole challenge exchange '
+Typeage = namedtuple("Tierage", 'oobi witness delegation group query registry credential endrole challenge exchange submit '
                                 'done')
 
 OpTypes = Typeage(oobi="oobi", witness='witness', delegation='delegation', group='group', query='query',
                   registry='registry', credential='credential', endrole='endrole', challenge='challenge',
-                  exchange='exchange', done='done')
+                  exchange='exchange', submit='submit', done='done')
 
 
 @dataclass_json
@@ -94,7 +94,7 @@ class Monitor:
 
     """
 
-    def __init__(self, hby, swain, counselor=None, registrar=None, exchanger=None, credentialer=None, opr=None,
+    def __init__(self, hby, swain, counselor=None, registrar=None, exchanger=None, credentialer=None, submitter=None, opr=None,
                  temp=False):
         """ Create long running operation monitor
 
@@ -110,6 +110,7 @@ def __init__(self, hby, swain, counselor=None, registrar=None, exchanger=None, c
         self.registrar = registrar
         self.exchanger = exchanger
         self.credentialer = credentialer
+        self.submitter = submitter
         self.opr = opr if opr is not None else Operator(name=hby.name, temp=temp)
 
     def submit(self, oid, typ, metadata=None):
@@ -412,6 +413,26 @@ def status(self, op):
             else:
                 operation.done = False
 
+        elif op.type in (OpTypes.submit,):
+            kever = self.hby.kevers[op.oid]
+            if kever and len(self.submitter.submits) == 0 and len(self.submitter.doers) == 0:
+                operation.done = True
+                operation.response = asdict(kever.state())
+            else:
+                start = helping.fromIso8601(op.start)
+                dtnow = helping.nowUTC()
+                if (dtnow - start) > datetime.timedelta(
+                    seconds=eventing.Kevery.TimeoutPWE
+                ):
+                    operation.done = True
+                    operation.error = Status(
+                        code=408,  # Using HTTP error codes here for lack of a better alternative
+                        message=f"long running {op.type} for {op.oid} operation timed out before "
+                        f"receiving sufficient witness receipts",
+                    )
+                else:
+                    operation.done = False
+
         elif op.type in (OpTypes.done, ):
             operation.done = True
             operation.response = op.metadata["response"]

src/keria/testing/testing_helper.py

@@ -428,7 +428,7 @@ def createRotate(aid, salt, signers, pidx, ridx, kidx, wits, toad):
         salter = core.Salter(raw=salt)
         creator = keeping.SaltyCreator(salt=salter.qb64, stem="signify:aid", tier=coring.Tiers.low)
         encrypter = core.Encrypter(verkey=signers[0].verfer.qb64)
-        sxlt = encrypter.encrypt(salter.qb64).qb64
+        sxlt = encrypter.encrypt(ser=salter.qb64).qb64
 
         rsigners = creator.create(pidx=pidx, ridx=ridx, tier=coring.Tiers.low, temp=False, count=1)
         rnsigners = creator.create(pidx=pidx, ridx=ridx+1, tier=coring.Tiers.low, temp=False, count=1)
@@ -468,7 +468,7 @@ def createAid(client, name, salt, wits=None, toad="0", delpre=None):
 
         salter = core.Salter(raw=salt)
         encrypter = core.Encrypter(verkey=signers[0].verfer.qb64)
-        sxlt = encrypter.encrypt(salter.qb64).qb64
+        sxlt = encrypter.encrypt(ser=salter.qb64).qb64
 
         sigers = [signer.sign(ser=serder.raw, index=0).qb64 for signer in signers]
 
@@ -599,6 +599,26 @@ def mockNowIso8601():
     def mockRandomNonce():
         return "A9XfpxIl1LcIkMhUSCCC8fgvkuX8gG9xK3SM-S8a8Y_U"
 
+    @staticmethod
+    def witnessMsg(hab, msg, sn, witHabs):
+        rctMsgs = []
+        for i, witHab in enumerate(witHabs):
+            kvy = witHab.kvy
+            witHab.psr.parse(ims=bytearray(msg), kvy=kvy, local=True)
+            # accepted event with cam sigs since own witness
+            assert kvy.kevers[hab.pre].sn == sn
+            assert len(kvy.cues) >= 1  # at least queued receipt cue
+            # better to find receipt cue in cues exactly
+            rctMsg = witHab.processCues(kvy.cues)  # process cue returns rct msg
+            assert len(rctMsg) > len(msg)
+            rctMsgs.append(rctMsg)
+
+        for rMsg in rctMsgs:  # process rct msgs from all witnesses
+            hab.psr.parse(ims=bytearray(rMsg), kvy=hab.kvy, local=True)
+        for whab in witHabs:
+            assert whab.pre in hab.kvy.kevers
+
+        return rctMsgs
 
 class Issuer:
     LE = "ENTAoj2oNBFpaniRswwPcca9W1ElEeH2V7ahw68HV4G5"